From 67d206dbd0fe3c084cf916c7ee77d4074c9a268d Mon Sep 17 00:00:00 2001
From: fhanik <fhanik@13f79535-47bb-0310-9956-ffa450edef68>
Date: Tue, 19 Feb 2008 20:45:54 +0000
Subject: [PATCH] more UDP impl

git-svn-id: https://svn.apache.org/repos/asf/tomcat/trunk@629223 13f79535-47bb-0310-9956-ffa450edef68
---
 .../apache/catalina/tribes/ChannelReceiver.java    |   2 +
 .../tribes/membership/McastServiceImpl.java        |   5 +-
 .../catalina/tribes/transport/ReceiverBase.java    |   1 -
 .../catalina/tribes/transport/nio/NioReceiver.java |   2 +-
 .../tribes/transport/nio/NioReplicationTask.java   |  10 +-
 .../catalina/tribes/transport/nio/NioSender.java   |  31 ++-
 .../tribes/test/channel/TestDataIntegrity.java     |  26 +--
 .../tribes/test/channel/TestUdpPackages.java       | 217 +++++++++++++++++++++
 8 files changed, 266 insertions(+), 28 deletions(-)
 create mode 100644 test/org/apache/catalina/tribes/test/channel/TestUdpPackages.java

diff --git a/java/org/apache/catalina/tribes/ChannelReceiver.java b/java/org/apache/catalina/tribes/ChannelReceiver.java
index f52c039e5..32c53e497 100644
--- a/java/org/apache/catalina/tribes/ChannelReceiver.java
+++ b/java/org/apache/catalina/tribes/ChannelReceiver.java
@@ -27,6 +27,8 @@ package org.apache.catalina.tribes;
  * @version $Revision$, $Date$
  */
 public interface ChannelReceiver extends Heartbeat {
+    public static final int MAX_UDP_SIZE = 65535;
+    
     /**
      * Start listening for incoming messages on the host/port
      * @throws java.io.IOException
diff --git a/java/org/apache/catalina/tribes/membership/McastServiceImpl.java b/java/org/apache/catalina/tribes/membership/McastServiceImpl.java
index c17affa1c..fd410bde7 100644
--- a/java/org/apache/catalina/tribes/membership/McastServiceImpl.java
+++ b/java/org/apache/catalina/tribes/membership/McastServiceImpl.java
@@ -335,7 +335,10 @@ public class McastServiceImpl
                         }
                     };
                 } //end if
-                if ( t != null ) t.start();
+                if ( t != null ) {
+                    t.setDaemon(true);
+                    t.start();
+                }
             }
         } catch (SocketTimeoutException x ) { 
             //do nothing, this is normal, we don't want to block forever
diff --git a/java/org/apache/catalina/tribes/transport/ReceiverBase.java b/java/org/apache/catalina/tribes/transport/ReceiverBase.java
index 3bc54a26d..06d4217c3 100644
--- a/java/org/apache/catalina/tribes/transport/ReceiverBase.java
+++ b/java/org/apache/catalina/tribes/transport/ReceiverBase.java
@@ -46,7 +46,6 @@ public abstract class ReceiverBase implements ChannelReceiver, ListenCallback, R
 
     public static final int OPTION_DIRECT_BUFFER = 0x0004;
 
-
     protected static org.apache.juli.logging.Log log = org.apache.juli.logging.LogFactory.getLog(ReceiverBase.class);
 
     private MessageListener listener;
diff --git a/java/org/apache/catalina/tribes/transport/nio/NioReceiver.java b/java/org/apache/catalina/tribes/transport/nio/NioReceiver.java
index b9ef3ba14..de527633e 100644
--- a/java/org/apache/catalina/tribes/transport/nio/NioReceiver.java
+++ b/java/org/apache/catalina/tribes/transport/nio/NioReceiver.java
@@ -249,7 +249,7 @@ public class NioReceiver extends ReceiverBase implements Runnable, ChannelReceiv
 
         setListen(true);
         if (selector!=null && datagramChannel!=null) {
-            ObjectReader oreader = new ObjectReader(65535); //max size for a datagram packet
+            ObjectReader oreader = new ObjectReader(MAX_UDP_SIZE); //max size for a datagram packet
             registerChannel(selector,datagramChannel,SelectionKey.OP_READ,oreader);
         }
 
diff --git a/java/org/apache/catalina/tribes/transport/nio/NioReplicationTask.java b/java/org/apache/catalina/tribes/transport/nio/NioReplicationTask.java
index 4fea62b0d..6e3707027 100644
--- a/java/org/apache/catalina/tribes/transport/nio/NioReplicationTask.java
+++ b/java/org/apache/catalina/tribes/transport/nio/NioReplicationTask.java
@@ -33,6 +33,8 @@ import org.apache.catalina.tribes.io.ListenCallback;
 import org.apache.catalina.tribes.io.ChannelData;
 import org.apache.catalina.tribes.io.BufferPool;
 import java.nio.channels.CancelledKeyException;
+
+import org.apache.catalina.tribes.ChannelReceiver;
 import org.apache.catalina.tribes.UniqueId;
 import org.apache.catalina.tribes.RemoteProcessException;
 import org.apache.catalina.tribes.util.Logs;
@@ -68,10 +70,14 @@ public class NioReplicationTask extends AbstractRxTask {
     // loop forever waiting for work to do
     public synchronized void run() {
         if ( buffer == null ) {
+            int size = getRxBufSize();
+            if (key.channel() instanceof DatagramChannel) {
+                size = ChannelReceiver.MAX_UDP_SIZE;
+            }
             if ( (getOptions() & OPTION_DIRECT_BUFFER) == OPTION_DIRECT_BUFFER) {
-                buffer = ByteBuffer.allocateDirect(getRxBufSize());
+                buffer = ByteBuffer.allocateDirect(size);
             } else {
-                buffer = ByteBuffer.allocate(getRxBufSize());
+                buffer = ByteBuffer.allocate(size);
             }
         } else {
             buffer.clear();
diff --git a/java/org/apache/catalina/tribes/transport/nio/NioSender.java b/java/org/apache/catalina/tribes/transport/nio/NioSender.java
index fe21cbf45..b8e0479ef 100644
--- a/java/org/apache/catalina/tribes/transport/nio/NioSender.java
+++ b/java/org/apache/catalina/tribes/transport/nio/NioSender.java
@@ -137,16 +137,24 @@ public class NioSender extends AbstractSender implements DataSender{
         connecting = false;
         setRequestCount(0);
         setConnectTime(System.currentTimeMillis());
-        socketChannel.socket().setSendBufferSize(getTxBufSize());
-        socketChannel.socket().setReceiveBufferSize(getRxBufSize());
-        socketChannel.socket().setSoTimeout((int)getTimeout());
-        socketChannel.socket().setSoLinger(getSoLingerOn(),getSoLingerOn()?getSoLingerTime():0);
-        socketChannel.socket().setTcpNoDelay(getTcpNoDelay());
-        socketChannel.socket().setKeepAlive(getSoKeepAlive());
-        socketChannel.socket().setReuseAddress(getSoReuseAddress());
-        socketChannel.socket().setOOBInline(getOoBInline());
-        socketChannel.socket().setSoLinger(getSoLingerOn(),getSoLingerTime());
-        socketChannel.socket().setTrafficClass(getSoTrafficClass());
+        if (socketChannel!=null) {
+            socketChannel.socket().setSendBufferSize(getTxBufSize());
+            socketChannel.socket().setReceiveBufferSize(getRxBufSize());
+            socketChannel.socket().setSoTimeout((int)getTimeout());
+            socketChannel.socket().setSoLinger(getSoLingerOn(),getSoLingerOn()?getSoLingerTime():0);
+            socketChannel.socket().setTcpNoDelay(getTcpNoDelay());
+            socketChannel.socket().setKeepAlive(getSoKeepAlive());
+            socketChannel.socket().setReuseAddress(getSoReuseAddress());
+            socketChannel.socket().setOOBInline(getOoBInline());
+            socketChannel.socket().setSoLinger(getSoLingerOn(),getSoLingerTime());
+            socketChannel.socket().setTrafficClass(getSoTrafficClass());
+        } else if (dataChannel!=null) {
+            dataChannel.socket().setSendBufferSize(getTxBufSize());
+            dataChannel.socket().setReceiveBufferSize(getRxBufSize());
+            dataChannel.socket().setSoTimeout((int)getTimeout());
+            dataChannel.socket().setReuseAddress(getSoReuseAddress());
+            dataChannel.socket().setTrafficClass(getSoTrafficClass());
+        }
     }
 
 
@@ -224,6 +232,9 @@ public class NioSender extends AbstractSender implements DataSender{
             dataChannel = DatagramChannel.open();
             dataChannel.configureBlocking(false);
             dataChannel.connect(daddr);
+            completeConnect();
+            dataChannel.register(getSelector(),SelectionKey.OP_WRITE, this);
+            
         } else {
             InetSocketAddress addr = new InetSocketAddress(getAddress(),getPort());
             if ( socketChannel != null ) throw new IOException("Socket channel has already been established. Connection might be in progress.");
diff --git a/test/org/apache/catalina/tribes/test/channel/TestDataIntegrity.java b/test/org/apache/catalina/tribes/test/channel/TestDataIntegrity.java
index 07b77bccc..2dfcd6e78 100644
--- a/test/org/apache/catalina/tribes/test/channel/TestDataIntegrity.java
+++ b/test/org/apache/catalina/tribes/test/channel/TestDataIntegrity.java
@@ -28,12 +28,12 @@ import org.apache.catalina.tribes.group.interceptors.MessageDispatchInterceptor;
 import org.apache.catalina.tribes.group.interceptors.MessageDispatch15Interceptor;
 
 /**
- * <p>Title: </p> 
- * 
- * <p>Description: </p> 
- * 
+ * <p>Title: </p>
+ *
+ * <p>Description: </p>
+ *
  * <p>Company: </p>
- * 
+ *
  * @author not attributable
  * @version 1.0
  */
@@ -61,7 +61,7 @@ public class TestDataIntegrity extends TestCase {
         channel1.stop(GroupChannel.DEFAULT);
         channel2.stop(GroupChannel.DEFAULT);
     }
-    
+
     public void testDataSendNO_ACK() throws Exception {
         System.err.println("Starting NO_ACK");
         Thread[] threads = new Thread[threadCount];
@@ -89,7 +89,7 @@ public class TestDataIntegrity extends TestCase {
         System.err.println("Finished NO_ACK ["+listener1.count+"]");
         assertEquals("Checking success messages.",msgCount*threadCount,listener1.count);
     }
-    
+
     public void testDataSendASYNCM() throws Exception {
             System.err.println("Starting ASYNC MULTI THREAD");
             Thread[] threads = new Thread[threadCount];
@@ -113,7 +113,7 @@ public class TestDataIntegrity extends TestCase {
             for (int x=0; x<threads.length; x++ ) { threads[x].join();}
             //sleep for 50 sec, let the other messages in
             long start = System.currentTimeMillis();
-            while ( (System.currentTimeMillis()-start)<15000 && msgCount*threadCount!=listener1.count) Thread.sleep(500);
+            while ( (System.currentTimeMillis()-start)<25000 && msgCount*threadCount!=listener1.count) Thread.sleep(500);
             System.err.println("Finished ASYNC MULTI THREAD ["+listener1.count+"]");
             assertEquals("Checking success messages.",msgCount*threadCount,listener1.count);
     }
@@ -148,7 +148,7 @@ public class TestDataIntegrity extends TestCase {
         public boolean accept(Serializable s, Member m) {
             return (s instanceof Data);
         }
-        
+
         public void messageReceived(Serializable s, Member m) {
             Data d = (Data)s;
             if ( !Data.verify(d) ) {
@@ -161,7 +161,7 @@ public class TestDataIntegrity extends TestCase {
             }
         }
     }
-    
+
     public static class Data implements Serializable {
         public int length;
         public byte[] data;
@@ -178,14 +178,14 @@ public class TestDataIntegrity extends TestCase {
             Arrays.fill(d.data,d.key);
             return d;
         }
-        
+
         public static boolean verify(Data d) {
             boolean result = (d.length == d.data.length);
             for ( int i=0; result && (i<d.data.length); i++ ) result = result && d.data[i] == d.key;
             return result;
         }
     }
-    
-    
+
+
 
 }
diff --git a/test/org/apache/catalina/tribes/test/channel/TestUdpPackages.java b/test/org/apache/catalina/tribes/test/channel/TestUdpPackages.java
new file mode 100644
index 000000000..c159d28a0
--- /dev/null
+++ b/test/org/apache/catalina/tribes/test/channel/TestUdpPackages.java
@@ -0,0 +1,217 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.test.channel;
+
+import junit.framework.TestCase;
+import java.io.Serializable;
+import java.util.Random;
+import java.util.Arrays;
+
+import org.apache.catalina.tribes.Channel;
+import org.apache.catalina.tribes.ChannelListener;
+import org.apache.catalina.tribes.ChannelReceiver;
+import org.apache.catalina.tribes.Member;
+import org.apache.catalina.tribes.group.GroupChannel;
+import org.apache.catalina.tribes.test.channel.TestDataIntegrity.Listener;
+import org.apache.catalina.tribes.transport.AbstractSender;
+import org.apache.catalina.tribes.transport.ReceiverBase;
+import org.apache.catalina.tribes.transport.ReplicationTransmitter;
+import org.apache.catalina.tribes.group.interceptors.MessageDispatchInterceptor;
+import org.apache.catalina.tribes.group.interceptors.MessageDispatch15Interceptor;
+import org.apache.catalina.tribes.group.interceptors.ThroughputInterceptor;
+
+/**
+ */
+public class TestUdpPackages extends TestCase {
+    int msgCount = 500;
+    int threadCount = 20;
+    GroupChannel channel1;
+    GroupChannel channel2;
+    Listener listener1;
+    int threadCounter = 0;
+    protected void setUp() throws Exception {
+        super.setUp();
+        channel1 = new GroupChannel();
+        channel1.addInterceptor(new MessageDispatch15Interceptor());
+        channel2 = new GroupChannel();
+        channel2.addInterceptor(new MessageDispatch15Interceptor());
+        ThroughputInterceptor tint = new ThroughputInterceptor();
+        tint.setInterval(500);
+        ThroughputInterceptor tint2 = new ThroughputInterceptor();
+        tint2.setInterval(500);
+        channel1.addInterceptor(tint);
+        channel2.addInterceptor(tint2);
+        listener1 = new Listener();
+        ReceiverBase rb1 = (ReceiverBase)channel1.getChannelReceiver();
+        ReceiverBase rb2 = (ReceiverBase)channel2.getChannelReceiver();
+        rb1.setUdpPort(50000);
+        rb2.setUdpPort(50000);
+        channel2.addChannelListener(listener1);
+        channel1.start(GroupChannel.DEFAULT);
+        channel2.start(GroupChannel.DEFAULT);
+    }
+
+    protected void tearDown() throws Exception {
+        super.tearDown();
+        channel1.stop(GroupChannel.DEFAULT);
+        channel2.stop(GroupChannel.DEFAULT);
+    }
+
+    public void testSingleDataSendNO_ACK() throws Exception {
+        AbstractSender s1 =(AbstractSender) ((ReplicationTransmitter)channel1.getChannelSender()).getTransport();
+        AbstractSender s2 =(AbstractSender) ((ReplicationTransmitter)channel2.getChannelSender()).getTransport();
+        s1.setTimeout(Long.MAX_VALUE); //for debugging
+        s2.setTimeout(Long.MAX_VALUE); //for debugging
+        
+        System.err.println("Starting Single package NO_ACK");
+        channel1.send(new Member[] {channel2.getLocalMember(false)}, Data.createRandomData(1024),Channel.SEND_OPTIONS_UDP);
+        Thread.sleep(500);
+        System.err.println("Finished Single package NO_ACK ["+listener1.count+"]");
+        assertEquals("Checking success messages.",1,listener1.count);
+    }
+
+    
+    public void testDataSendNO_ACK() throws Exception {
+        System.err.println("Starting NO_ACK");
+        Thread[] threads = new Thread[threadCount];
+        for (int x=0; x<threads.length; x++ ) {
+            threads[x] = new Thread() {
+                public void run() {
+                    try {
+                        long start = System.currentTimeMillis();
+                        for (int i = 0; i < msgCount; i++) channel1.send(new Member[] {channel2.getLocalMember(false)}, Data.createRandomData(1024),Channel.SEND_OPTIONS_UDP);
+                        System.out.println("Thread["+this.getName()+"] sent "+msgCount+" messages in "+(System.currentTimeMillis()-start)+" ms.");
+                    }catch ( Exception x ) {
+                        x.printStackTrace();
+                        return;
+                    } finally {
+                        threadCounter++;
+                    }
+                }
+            };
+        } 
+        for (int x=0; x<threads.length; x++ ) { threads[x].start();}
+        for (int x=0; x<threads.length; x++ ) { threads[x].join();}
+        //sleep for 50 sec, let the other messages in
+        long start = System.currentTimeMillis();
+        while ( (System.currentTimeMillis()-start)<25000 && msgCount*threadCount!=listener1.count) Thread.sleep(500);
+        System.err.println("Finished NO_ACK ["+listener1.count+"]");
+        assertEquals("Checking success messages.",msgCount*threadCount,listener1.count);
+    }
+
+    public void testDataSendASYNCM() throws Exception {
+            System.err.println("Starting ASYNC MULTI THREAD");
+            Thread[] threads = new Thread[threadCount];
+            for (int x=0; x<threads.length; x++ ) {
+                threads[x] = new Thread() {
+                    public void run() {
+                        try {
+                            long start = System.currentTimeMillis();
+                            for (int i = 0; i < msgCount; i++) channel1.send(new Member[] {channel2.getLocalMember(false)}, Data.createRandomData(1024),GroupChannel.SEND_OPTIONS_ASYNCHRONOUS|Channel.SEND_OPTIONS_UDP);
+                            System.out.println("Thread["+this.getName()+"] sent "+msgCount+" messages in "+(System.currentTimeMillis()-start)+" ms.");
+                        }catch ( Exception x ) {
+                            x.printStackTrace();
+                            return;
+                        } finally {
+                            threadCounter++;
+                        }
+                    }
+                };
+            }
+            for (int x=0; x<threads.length; x++ ) { threads[x].start();}
+            for (int x=0; x<threads.length; x++ ) { threads[x].join();}
+            //sleep for 50 sec, let the other messages in
+            long start = System.currentTimeMillis();
+            while ( (System.currentTimeMillis()-start)<25000 && msgCount*threadCount!=listener1.count) Thread.sleep(500);
+            System.err.println("Finished ASYNC MULTI THREAD ["+listener1.count+"]");
+            assertEquals("Checking success messages.",msgCount*threadCount,listener1.count);
+    }
+    public void testDataSendASYNC() throws Exception {
+        System.err.println("Starting ASYNC");
+        for (int i=0; i<msgCount; i++) channel1.send(new Member[] {channel2.getLocalMember(false)},Data.createRandomData(1024),GroupChannel.SEND_OPTIONS_ASYNCHRONOUS|Channel.SEND_OPTIONS_UDP);
+        //sleep for 50 sec, let the other messages in
+        long start = System.currentTimeMillis();
+        while ( (System.currentTimeMillis()-start)<5000 && msgCount!=listener1.count) Thread.sleep(500);
+        System.err.println("Finished ASYNC");
+        assertEquals("Checking success messages.",msgCount,listener1.count);
+    }
+
+    public void testDataSendACK() throws Exception {
+        System.err.println("Starting ACK");
+        for (int i=0; i<msgCount; i++) channel1.send(new Member[] {channel2.getLocalMember(false)},Data.createRandomData(1024),GroupChannel.SEND_OPTIONS_USE_ACK|Channel.SEND_OPTIONS_UDP);
+        Thread.sleep(250);
+        System.err.println("Finished ACK");
+        assertEquals("Checking success messages.",msgCount,listener1.count);
+    }
+
+    public void testDataSendSYNCACK() throws Exception {
+        System.err.println("Starting SYNC_ACK");
+        for (int i=0; i<msgCount; i++) channel1.send(new Member[] {channel2.getLocalMember(false)},Data.createRandomData(1024),GroupChannel.SEND_OPTIONS_SYNCHRONIZED_ACK|GroupChannel.SEND_OPTIONS_USE_ACK|Channel.SEND_OPTIONS_UDP);
+        Thread.sleep(250);
+        System.err.println("Finished SYNC_ACK");
+        assertEquals("Checking success messages.",msgCount,listener1.count);
+    }
+
+    public static class Listener implements ChannelListener {
+        long count = 0;
+        public boolean accept(Serializable s, Member m) {
+            return (s instanceof Data);
+        }
+
+        public void messageReceived(Serializable s, Member m) {
+            Data d = (Data)s;
+            if ( !Data.verify(d) ) {
+                System.err.println("ERROR");
+            } else {
+                count++;
+                if ((count %1000) ==0 ) {
+                    System.err.println("SUCCESS:"+count);
+                }
+            }
+        }
+    }
+
+    public static class Data implements Serializable {
+        public int length;
+        public byte[] data;
+        public byte key;
+        public static Random r = new Random(System.currentTimeMillis());
+        public static Data createRandomData() {
+            return createRandomData(ChannelReceiver.MAX_UDP_SIZE);
+        }
+        public static Data createRandomData(int size) {
+            int i = r.nextInt();
+            i = ( i % 127 );
+            int length = Math.abs(r.nextInt() % size);
+            Data d = new Data();
+            d.length = length;
+            d.key = (byte)i;
+            d.data = new byte[length];
+            Arrays.fill(d.data,d.key);
+            return d;
+        }
+
+        public static boolean verify(Data d) {
+            boolean result = (d.length == d.data.length);
+            for ( int i=0; result && (i<d.data.length); i++ ) result = result && d.data[i] == d.key;
+            return result;
+        }
+    }
+
+
+
+}
-- 
2.11.0