path: root/simple/simple-transport/src/test/java/org/simpleframework/transport/TransportTest.java

package org.simpleframework.transport;

import java.io.BufferedInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.LinkedList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicLong;

import junit.framework.TestCase;

import org.simpleframework.transport.reactor.ExecutorReactor;
import org.simpleframework.transport.reactor.Reactor;

/**
 * Measure the performance of the transports to ensure that the perform 
 * well and that they send the correct sequence of bytes and that the
 * blocks sent are in the correct order. This also performs a comparison
 * with direct socket output streams to ensure there is a reasonable
 * performance difference.  
 * 
 * @author Niall Gallagher 
 */
public class TransportTest extends TestCase {

   private static final int REPEAT = 1000;

   public void testTransport() throws Exception {
      testTransport(REPEAT);
   }

   public void testTransport(int repeat) throws Exception {    
      for(int i = 1; i < 7; i++) { // just do some random sizes 
         testTransport(i, 100);
      }
      for(int i = 4092; i < 4102; i++) {
         testTransport(i, 100);
      }
      for(int i = 8190; i < 8200; i++) {
         testTransport(i, 100);
      }
      for(int i = 11282; i < 11284; i++) {
         testTransport(i, 1000);
      }
      for(int i = 204800; i < 204805; i++) {
         testTransport(i, 1000);
      }
      testTransport(16, repeat);
      testTransport(64, repeat);
      testTransport(256, repeat);
      testTransport(1024, repeat);
      testTransport(2048, repeat);
      testTransport(4096, repeat); 
      testTransport(4098, repeat);
      testTransport(8192, repeat);
      testTransport(8197, repeat);
   }

   // Test blocking transport        
   private void testTransport(int size, int repeat) throws Exception {
     // ThreadDumper dumper = new ThreadDumper();
      SocketConsumer consumer = new SocketConsumer(size, repeat);
      SocketAddress address = new InetSocketAddress("localhost", consumer.getPort());
      SocketChannel channel = SocketChannel.open();
      channel.configureBlocking(false); // underlying socket must be non-blocking
      channel.connect(address);

      while(!channel.finishConnect()) { // wait to finish connection
         Thread.sleep(10);
      };
      ExecutorService executor = Executors.newFixedThreadPool(20);
      Reactor reactor = new ExecutorReactor(executor);
     // Transport transport = new SocketTransport(channel, reactor, 2, 3);//XXX bug
      MockSocket pipeline = new MockSocket(channel);
      Transport transport = new SocketTransport(pipeline, reactor, 8192);
      OutputStream out = new TransportOutputStream(transport);

     // dumper.start();
      testOutputStream(consumer, out, size, repeat);

      out.close();
      executor.shutdown();
      channel.close();
      reactor.stop();
    //  dumper.kill();
      Thread.sleep(100);
   }

   public void s_testSocket() throws Exception {
      s_testSocket(REPEAT);
   }

   public void s_testSocket(int repeat) throws Exception {
      testSocket(16, repeat);
      testSocket(64, repeat);
      testSocket(256, repeat);
      testSocket(1024, repeat);
      testSocket(2048, repeat);
      testSocket(4098, repeat);
      testSocket(8192, repeat);
   }

   // Test blocking socket
   private void testSocket(int size, int repeat) throws Exception {
     // ThreadDumper dumper = new ThreadDumper();
      SocketConsumer consumer = new SocketConsumer(size, repeat);
      Socket socket = new Socket("localhost", consumer.getPort());
      OutputStream out = socket.getOutputStream();

      //dumper.start();
      testOutputStream(consumer, out, size, repeat);

      out.close();
      socket.close();
      //dumper.kill();
      Thread.sleep(100);
   }

   private class AlpahbetIterator {

      private byte[] alphabet = "abcdefghijklmnopqstuvwxyz".getBytes();

      private int off;

      public byte next() {
         if(off == alphabet.length) {
            off = 0;
         }
         return alphabet[off++];
      }

      public void reset() {
         off = 0;
      }
   }

   private void testOutputStream(SocketConsumer consumer, OutputStream out, int size, int repeat) throws Exception {
      byte[] block = new byte[size]; // write size
      AlpahbetIterator it = new AlpahbetIterator(); // write known data   

      for(int i = 1; i < block.length; i++) {
         block[i] = it.next();
      }
      AtomicLong count = new AtomicLong();
      PerformanceMonitor monitor = new PerformanceMonitor(consumer, count, out.getClass().getSimpleName(), size);

      for(int i = 0; i < repeat; i++) {
         block[0] = (byte) i; // mark the first byte in the block to be sure we get blocks in sequence
         //System.err.println("["+i+"]"+new String(block,"ISO-8859-1"));
         out.write(block);  // manipulation of the underlying buffer is taking place when the compact is invoked, this is causing major problems as the next packet will be out of sequence
         count.addAndGet(block.length);
      }
      Thread.sleep(2000); // wait for all bytes to flush through to consumer
      monitor.kill();
   }

   private class PerformanceMonitor extends Thread {
      private AtomicLong count;

      private volatile boolean dead;
      
      private SocketConsumer consumer;

      private String name;

      private int size;

      public PerformanceMonitor(SocketConsumer consumer, AtomicLong count, String name, int size) {
         this.consumer = consumer;
         this.count = count;
         this.name = name;
         this.size = size;
         this.start();
      }

      public void run() {
         int second = 0;
         while(!dead) {
            try {
               long octets = count.longValue();
               System.out.printf("%s,%s,%s,%s,%s%n", name, size, second++, octets, consumer.getWindow());
               Thread.sleep(1000);
            } catch(Exception e) {
               e.printStackTrace();
            }
         }
      }

      public void kill() throws Exception {
         dead = true;
      }
   }

   private class SocketConsumer extends Thread {

      private ServerSocket server;

      private Window window;
      
      private long repeat;

      private long size;

      public SocketConsumer(int size, int repeat) throws Exception {
         this.window = new Window(20);
         this.server = getSocket();
         this.repeat = repeat;
         this.size = size;
         this.start();
      }
      
      public int getPort() {
         return server.getLocalPort();
      }
      
      public String getWindow() {
         return window.toString();
      }
      
      private ServerSocket getSocket() throws Exception {
         // Scan the ephemeral port range
         for(int i = 2000; i < 10000; i++) { // keep trying to grab the socket 
            try {
               ServerSocket socket = new ServerSocket(i);
               System.out.println("port=["+socket.getLocalPort()+"]");
               return socket;
            } catch(Exception e) {
               Thread.sleep(200);
            }
         }
         // Scan a second time for good measure, maybe something got freed up
         for(int i = 2000; i < 10000; i++) { // keep trying to grab the socket 
            try {
               ServerSocket socket = new ServerSocket(i);
               System.out.println("port=["+socket.getLocalPort()+"]");
               return socket;
            } catch(Exception e) {
               Thread.sleep(200);
            }
         }
         throw new IOException("Could not create a client socket");
      }

      public void run() {
         long count = 0;
         int windowOctet = 0;
         int expectWindowOctet = 0;

         try {
            Socket socket = server.accept();
            InputStream in = socket.getInputStream();
            InputStream source = new BufferedInputStream(in);
            AlpahbetIterator it = new AlpahbetIterator();

            scan: for(int i = 0; i < repeat; i++) {
               int octet = source.read(); // check first byte in the block to make sure its correct in sequence
               
               if(octet == -1) {
                  break scan;
               }
               count++; // we have read another byte
               windowOctet = octet & 0x000000ff;
               expectWindowOctet = i & 0x000000ff;
               
               if((byte) octet != (byte) i) {
                  throw new Exception("Wrong sequence of blocks sent, was "
                        + (byte)octet + " should have been " + (byte)i + " count is "+count+" window is "+window+" compare "+explore(it, source, 5));
               }
               window.recieved(octet);
               
               for(int j = 1, k = 0; j < size; j++, k++) {
                  octet = source.read();

                  if(octet == -1) {
                     break scan;
                  }
                  byte next = it.next();
                   
                  if((byte) octet != next) {
                     throw new Exception("Invalid data received expected "+((byte)octet)+"("+((char)octet)+
                           ") but was "+next+"("+((char)next)+") total count is "+count+" block count is "+k+" window is expected "+
                           expectWindowOctet+"("+((char)expectWindowOctet)+")("+((byte)expectWindowOctet)+") got "+windowOctet+"("+
                           ((char)windowOctet)+")("+((byte)windowOctet)+") "+window+" compare "+explore(it, source, 5));
                  }
                  count++;
               }
               it.reset();
            }
         } catch(Throwable e) {
            e.printStackTrace();
         }
         if(count != size * repeat) {
            new Exception("Invalid number of bytes read, was " + count
                  + " should have been " + (size * repeat)).printStackTrace();
         }      
         try {
           // server.close();         
         }catch(Exception e) {
            e.printStackTrace();
         }      
      }
      
      private String explore(AlpahbetIterator it, InputStream source, int count) throws IOException {
         StringBuffer buf = new StringBuffer();
         buf.append("expected (");
         for(int i = 0; i < count; i++) {
            buf.append( (char)it.next() );
         }
         buf.append(") is (");
         for(int i = 0; i < count; i++) {
            buf.append(  (char)source.read()  );
         }
         buf.append(")");
         return buf.toString();
      }
   }
   

   private static class TransportOutputStream extends OutputStream {

      private Transport transport;

      public TransportOutputStream(Transport transport) {
         this.transport = transport;
      }

      public void write(int octet) throws IOException {
         byte[] data = new byte[] { (byte) octet };
         write(data);
      }

      public void write(byte[] data, int off, int len) throws IOException {
         try {
            ByteBuffer buffer = ByteBuffer.wrap(data, off, len);
            ByteBuffer safe = buffer.asReadOnlyBuffer();
            
            if(len > 0) {
               transport.write(safe);
            }
         } catch(Exception e) {
            e.printStackTrace();
            throw new IOException("Write failed");
         }
      }

      public void flush() throws IOException {
         try {
            transport.flush();
         } catch(Exception e) {
            e.printStackTrace();
            throw new IOException("Flush failed");
         }
      }

      public void close() throws IOException {
         try {
            transport.close();
         } catch(Exception e) {
            e.printStackTrace();
            throw new IOException("Close failed");
         }
      }

   }
   
   private static class Window {
      
      private final LinkedList<String> window;
      private final int size;
      
      public Window(int size) {
         this.window = new LinkedList<String>();
         this.size = size;
      }
      
      public synchronized void recieved(int sequence) {
         window.addLast(String.valueOf(sequence));
         
         if(window.size() > size) {
            window.removeFirst();
         }
      }
      
      public synchronized String toString() {
         StringBuilder builder = new StringBuilder("[");
         String delim = "";
         for(String b : window) {
            builder.append(delim).append(b);
            delim=", ";
         }
         builder.append("]");
         return builder.toString();
      }
   }

}