tomcat-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From fha...@apache.org
Subject svn commit: r382412 - in /tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes: io/XByteBuffer.java tcp/DataSender.java tcp/NioSender.java tcp/ParallelNioSender.java
Date Thu, 02 Mar 2006 15:32:46 GMT
Author: fhanik
Date: Thu Mar  2 07:32:45 2006
New Revision: 382412

URL: http://svn.apache.org/viewcvs?rev=382412&view=rev
Log:
Worked on the NIO datasender, state machine is complete, time to start working on parallelism

Modified:
    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/tcp/DataSender.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/NioSender.java
    tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/ParallelNioSender.java

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=382412&r1=382411&r2=382412&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
Thu Mar  2 07:32:45 2006
@@ -238,12 +238,16 @@
         return true;
     }
 
-    private void expand(int newcount) {
+    public void expand(int newcount) {
         //don't change the allocation strategy
         byte newbuf[] = new byte[Math.max(buf.length << 1, newcount)];
         System.arraycopy(buf, 0, newbuf, 0, bufSize);
         buf = newbuf;
     }
+    
+    public int getCapacity() {
+        return buf.length;
+    }
 
 
     /**
@@ -285,7 +289,7 @@
 
     /**
      * Method to check if a package exists in this byte buffer.
-     * @return - true if a complete package (header,compress,size,data,footer) exists within
the buffer
+     * @return - true if a complete package (header,options,size,data,footer) exists within
the buffer
      */
     public boolean doesPackageExist()  {
         return (countPackages()>0);
@@ -297,19 +301,24 @@
      * @param clearFromBuffer - if true, the package will be removed from the byte buffer
      * @return - returns the actual message bytes (header, compress,size and footer not included).
      */
-    public ClusterData extractPackage(boolean clearFromBuffer)
-            throws java.io.IOException {
+    public byte[] extractDataPackage(boolean clearFromBuffer) {
         int psize = countPackages();
         if (psize == 0) throw new java.lang.IllegalStateException("No package exists in XByteBuffer");
         int size = toInt(buf, START_DATA.length);
         byte[] data = new byte[size];
         System.arraycopy(buf, START_DATA.length + 4, data, 0, size);
-        ClusterData cdata = ClusterData.getDataFromPackage(data);
         if (clearFromBuffer) {
             int totalsize = START_DATA.length + 4 + size + END_DATA.length;
             bufSize = bufSize - totalsize;
             System.arraycopy(buf, totalsize, buf, 0, bufSize);
         }
+        return data;
+
+    }
+    
+    public ClusterData extractPackage(boolean clearFromBuffer) throws java.io.IOException
{
+        byte[] data = extractDataPackage(clearFromBuffer);
+        ClusterData cdata = ClusterData.getDataFromPackage(data);
         return cdata;
     }
     
@@ -324,20 +333,38 @@
         return createDataPackage(cdata.getDataPackage());
     }
     
-    public static byte[] createDataPackage(byte[] data) {
+    public static int getDataPackageLength(int datalength) {
         int length = 
             START_DATA.length + //header length
             4 + //data length indicator
-            data.length + //actual data length
+            datalength + //actual data length
             END_DATA.length; //footer length
+        return length;
+
+    }
+    
+    public static byte[] createDataPackage(byte[] data) {
+        int length = getDataPackageLength(data.length);
         byte[] result = new byte[length];
-        System.arraycopy(START_DATA, 0, result, 0, START_DATA.length);
-        System.arraycopy(toBytes(data.length), 0, result, START_DATA.length, 4);
-        System.arraycopy(data, 0, result, START_DATA.length + 4, data.length);
-        System.arraycopy(END_DATA, 0, result, START_DATA.length + 4 + data.length, END_DATA.length);
-        return result;
+        return createDataPackage(data,0,data.length,result,0);
+    }
+    
+    public static byte[] createDataPackage(byte[] data, int doff, int dlength, byte[] buffer,
int bufoff) {
+        if ( (buffer.length-bufoff) > getDataPackageLength(dlength) ) {
+            throw new ArrayIndexOutOfBoundsException("Unable to create data package, buffer
is too small.");
+        }
+        System.arraycopy(START_DATA, 0, buffer, bufoff, START_DATA.length);
+        System.arraycopy(toBytes(data.length), 0, buffer, bufoff+START_DATA.length, 4);
+        System.arraycopy(data, doff, buffer, bufoff+START_DATA.length + 4, dlength);
+        System.arraycopy(END_DATA, 0, buffer, bufoff+START_DATA.length + 4 + data.length,
END_DATA.length);
+        return buffer;
     }
 
+    public static void fillDataPackage(byte[] data, int doff, int dlength, XByteBuffer buf)
{
+        int pkglen = getDataPackageLength(dlength);
+        if ( buf.getCapacity() <  pkglen ) buf.expand(pkglen);
+        createDataPackage(data,doff,dlength,buf.getBytesDirect(),buf.getLength());
+    }
 
     /**
      * Convert four bytes to an int

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=382412&r1=382411&r2=382412&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
Thu Mar  2 07:32:45 2006
@@ -26,6 +26,7 @@
 import org.apache.catalina.tribes.io.ClusterData;
 import org.apache.catalina.tribes.io.XByteBuffer;
 import org.apache.catalina.util.StringManager;
+import java.util.Arrays;
 
 /**
  * Send cluster messages with only one socket. Ack and keep Alive Handling is
@@ -921,7 +922,7 @@
                 byte d = (byte)i;
                 ackbuf.append(d);
                 if (ackbuf.doesPackageExist() ) {
-                    ackReceived = true;
+                    ackReceived = Arrays.equals(ackbuf.extractDataPackage(true),Constants.ACK_DATA);
                     break;
                 }
                 i = socket.getInputStream().read();

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/NioSender.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/NioSender.java?rev=382412&r1=382411&r2=382412&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/NioSender.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/NioSender.java
Thu Mar  2 07:32:45 2006
@@ -26,6 +26,9 @@
 import java.nio.channels.SelectionKey;
 import java.nio.ByteBuffer;
 import org.apache.catalina.tribes.io.XByteBuffer;
+import org.apache.catalina.tribes.Member;
+import java.util.Arrays;
+import org.apache.catalina.tribes.io.ClusterData;
 
 /**
  * This class is NOT thread safe and should never be used with more than one thread at a
time
@@ -47,25 +50,29 @@
 
     
     protected long ackTimeout = 15000;
-    protected InetAddress address;
     protected String domain = "";
-    protected int port;
     protected boolean suspect = false;
     protected boolean connected = false;
     protected boolean waitForAck = false;
     protected int rxBufSize = 25188;
     protected int txBufSize = 43800;
     protected Selector selector;
-    
+    protected Member destination;
     
     protected SocketChannel socketChannel;
-    protected ByteBuffer buf = null;
+
+    /*
+     * STATE VARIABLES *
+     */
+    protected ByteBuffer readbuf = null;
     protected boolean direct = false;
-    protected ChannelMessage current = null;
+    protected byte[] current = null;
     protected int curPos=0;
     protected XByteBuffer ackbuf = new XByteBuffer(128,true);
+    protected int remaining = 0;
 
-    public NioSender() {
+    public NioSender(Member destination) {
+        this.destination = destination;
         
     }
     
@@ -76,7 +83,11 @@
             if ( socketChannel.finishConnect() ) {
                 //we connected, register ourselves for writing
                 this.connected = true;
-                key.interestOps(key.interestOps() | SelectionKey.OP_WRITE);
+                if ( current != null ) key.interestOps(key.interestOps() | SelectionKey.OP_WRITE);
+                return false;
+            } else  { 
+                //wait for the connection to finish
+                key.interestOps(key.interestOps() | SelectionKey.OP_CONNECT);
                 return false;
             }
         } else if ( key.isWritable() ) {
@@ -84,7 +95,7 @@
             if ( writecomplete ) {
                 //we are completed, should we read an ack?
                 if ( waitForAck ) key.interestOps(key.interestOps()|SelectionKey.OP_READ);
-                //if not, we are ready, setMessage will reregister us for write interest
+                //if not, we are ready, setMessage will reregister us for another write interest
                 else return true;
             } else {
                 //we are not complete, lets write some more
@@ -106,12 +117,19 @@
     protected boolean read(SelectionKey key) throws IOException {
         //if there is no message here, we are done
         if ( current == null ) return true;
-        int read = socketChannel.read(buf);
+        int read = socketChannel.read(readbuf);
         //end of stream
-        if ( read == -1 ) return true;
+        if ( read == -1 ) throw new IOException("Unable to receive an ack message.");
         //no data read
         else if ( read == 0 ) return false;
-        throw new UnsupportedOperationException();
+        readbuf.flip();
+        ackbuf.append(readbuf,read);
+        readbuf.clear();
+        if (ackbuf.doesPackageExist() ) {
+            return Arrays.equals(ackbuf.extractDataPackage(true),Constants.ACK_DATA);
+        } else {
+            return false;
+        }
     }
 
     
@@ -120,25 +138,20 @@
             throw new IOException("NioSender is not connected, this should not occur.");
         }
         if ( current != null ) {
-            int remaining = buf.remaining();
             if ( remaining > 0 ) {
-                //write the rest of the buffer
-                remaining -= socketChannel.write(buf);
-            }            
-            if ( remaining == 0 ) {
                 //weve written everything, or we are starting a new package
-                XByteBuffer msg = current.getMessage();
-                remaining = msg.getLength() - curPos;
-                buf.clear();
                 //protect against buffer overwrite
-                int length = Math.min(remaining,txBufSize);
-                buf.put(msg.getBytesDirect(),curPos,length);
-                
-                //if the entire message fits in the buffer
+                int length = current.length-curPos;
+                ByteBuffer writebuf = ByteBuffer.wrap(current,curPos,length);
+                int byteswritten = socketChannel.write(writebuf);
+                curPos += byteswritten;
+                remaining -= byteswritten;
+                //if the entire message was written from the buffer
                 //reset the position counter
-                curPos += length;
-                if ( curPos >= msg.getLength() ) curPos = 0;
-                remaining -= socketChannel.write(buf);
+                if ( curPos >= current.length ) {
+                    curPos = 0;
+                    remaining = 0;
+                }
             }
             //the write 
             return (remaining==0 && curPos == 0);
@@ -155,11 +168,12 @@
      */
     public synchronized void connect() throws IOException {
         if ( connected ) throw new IOException("NioSender is already in connected state.");
-        if ( buf == null ) {
-            if ( direct ) buf = ByteBuffer.allocateDirect(txBufSize);
-            else buf = ByteBuffer.allocate(txBufSize);
+        if ( readbuf == null ) {
+            readbuf = getReadBuffer();
+        } else {
+            readbuf.clear();
         }
-        InetSocketAddress addr = new InetSocketAddress(address,port);
+        InetSocketAddress addr = new InetSocketAddress(InetAddress.getByAddress(destination.getHost()),destination.getPort());
         if ( socketChannel != null ) throw new IOException("Socket channel has already been
established. Connection might be in progress.");
         socketChannel = SocketChannel.open();
         socketChannel.configureBlocking(false);
@@ -177,14 +191,29 @@
     public void disconnect() {
         try {
             this.connected = false;
-            if ( buf != null ) buf.clear();
             socketChannel.close();
             socketChannel = null;
-            curPos = 0;
         } catch ( Exception x ) {
             log.error("Unable to disconnect.",x);
+        } finally {
+            reset();
+        }
+
+    }
+    
+    public void reset() {
+        if ( connected && readbuf == null) {
+            readbuf = getReadBuffer();
         }
+        if ( readbuf != null ) readbuf.clear();
+        current = null;
+        curPos = 0;
+        ackbuf.clear();
+        remaining = 0;
+    }
 
+    private ByteBuffer getReadBuffer() {
+        return (direct?ByteBuffer.allocateDirect(rxBufSize):ByteBuffer.allocate(rxBufSize));
     }
     
     /**
@@ -195,14 +224,15 @@
     * @todo Implement this org.apache.catalina.tribes.tcp.IDataSender method
     */
    public synchronized void setMessage(ChannelMessage data) throws IOException {
-       this.current = data;
+       reset();
        if ( data != null ) {
-           if (!this.connected) {
-               connect();
-           } else {
+           current = XByteBuffer.createDataPackage((ClusterData)data);
+           remaining = current.length;
+           curPos = 0;
+           if (connected) {
                socketChannel.register(getSelector(), SelectionKey.OP_WRITE, this);
            }
-       }
+       } 
    }
 
 
@@ -226,35 +256,7 @@
         return this.ackTimeout;
     }
 
-    /**
-     * getAddress
-     *
-     * @return InetAddress
-     * @todo Implement this org.apache.catalina.tribes.tcp.IDataSender method
-     */
-    public InetAddress getAddress() {
-        return address;
-    }
-
-    /**
-     * getDomain
-     *
-     * @return String
-     * @todo Implement this org.apache.catalina.tribes.tcp.IDataSender method
-     */
-    public String getDomain() {
-        return domain;
-    }
-
-    /**
-     * getPort
-     *
-     * @return int
-     * @todo Implement this org.apache.catalina.tribes.tcp.IDataSender method
-     */
-    public int getPort() {
-        return port;
-    }
+    
 
     /**
      * getSuspect
@@ -303,36 +305,6 @@
      */
     public void setAckTimeout(long timeout) {
         this.ackTimeout = timeout;
-    }
-
-    /**
-     * setAddress
-     *
-     * @param address InetAddress
-     * @todo Implement this org.apache.catalina.tribes.tcp.IDataSender method
-     */
-    public void setAddress(InetAddress address) {
-        this.address = address;
-    }
-
-    /**
-     * setDomain
-     *
-     * @param domain String
-     * @todo Implement this org.apache.catalina.tribes.tcp.IDataSender method
-     */
-    public void setDomain(String domain) {
-        this.domain = domain;
-    }
-
-    /**
-     * setPort
-     *
-     * @param port int
-     * @todo Implement this org.apache.catalina.tribes.tcp.IDataSender method
-     */
-    public void setPort(int port) {
-        this.port = port;
     }
 
     /**

Modified: tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/ParallelNioSender.java
URL: http://svn.apache.org/viewcvs/tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/ParallelNioSender.java?rev=382412&r1=382411&r2=382412&view=diff
==============================================================================
--- tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/ParallelNioSender.java
(original)
+++ tomcat/container/tc5.5.x/modules/groupcom/src/share/org/apache/catalina/tribes/tcp/ParallelNioSender.java
Thu Mar  2 07:32:45 2006
@@ -17,13 +17,8 @@
 package org.apache.catalina.tribes.tcp;
 
 /**
- * <p>Title: </p>
- *
- * <p>Description: </p>
- *
- * <p>Copyright: Copyright (c) 2005</p>
- *
- * <p>Company: </p>
+ * A class that uses NIO to send data in parallel to several remote nodes.
+ * 
  *
  * @author Filip Hanik
  * @version 1.0



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


Mime
View raw message