    1   /***
    2    *
    3    * Copyright 2004 Hiram Chirino
    4    *
    5    *  Licensed under the Apache License, Version 2.0 (the "License");
    6    *  you may not use this file except in compliance with the License.
    7    *  You may obtain a copy of the License at
    8    *
    9    *     http://www.apache.org/licenses/LICENSE-2.0
    10   *
    11   *  Unless required by applicable law or agreed to in writing, software
    12   *  distributed under the License is distributed on an "AS IS" BASIS,
    13   *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    14   *  See the License for the specific language governing permissions and
    15   *  limitations under the License.
    16   */
    17  package org.activeio.net.benchmark;
    18
    19  import java.io.IOException;
    20  import java.lang.reflect.InvocationTargetException;
    21  import java.net.URI;
    22  import java.net.URISyntaxException;
    23  import java.util.HashMap;
    24
    25  import org.activeio.AcceptListener;
    26  import org.activeio.AsynchChannel;
    27  import org.activeio.AsynchChannelListener;
    28  import org.activeio.AsynchChannelServer;
    29  import org.activeio.Channel;
    30  import org.activeio.ChannelFactory;
    31  import org.activeio.Packet;
    32  import org.activeio.adapter.SynchToAsynchChannelAdapter;
    33  import org.activeio.packet.EOSPacket;
    34  import org.activeio.stats.CountStatisticImpl;
    35  import org.apache.commons.beanutils.BeanUtils;
    36
    37  import EDU.oswego.cs.dl.util.concurrent.Latch;
    38
    39  /***
    40   * Implements a simple tcp echo server for use in benchmarking
    41   * activeio channel implementations.
    42   *
    43   * @version $Revision$
    44   */
    45  public class Server implements Runnable, AcceptListener {
    46
    47      private URI url;
    48      private Latch shutdownLatch;
    49      private long requestDelay = 0;
    50      private long sampleInterval = 1000;
    51
    52      private final CountStatisticImpl activeConnectionsCounter = new CountStatisticImpl("activeConnectionsCounter","The number of active connection attached to the server.");
    53      private final CountStatisticImpl echoedBytesCounter = new CountStatisticImpl("echoedBytesCounter","The number of bytes that have been echoed by the server.");
    54
    55      public static void main(String[] args) throws URISyntaxException, IllegalAccessException, InvocationTargetException {
    56
    57          Server server = new Server();
    58
    59          HashMap options = new HashMap();
    60          for( int i=0; i < args.length; i++ ) {
    61
    62              String option = args[i];
    63              if( !option.startsWith("-") || option.length()<2 || i+1 >= args.length ) {
    64                  System.out.println("Invalid usage.");
    65                  return;
    66              }
    67
    68              option = option.substring(1);
    69              options.put(option, args[++i]);
    70          }
    71          BeanUtils.populate(server, options);
    72
    73          System.out.println();
    74          System.out.println("Server starting with the following options: ");
    75          System.out.println(" url="+server.getUrl());
    76          System.out.println(" sampleInterval="+server.getSampleInterval());
    77          System.out.println(" requestDelay="+server.getRequestDelay());
    78          System.out.println();
    79          server.run();
    80
    81      }
    82
    83      private void printSampleData() {
    84          long now = System.currentTimeMillis();
    85          float runDuration = (now - activeConnectionsCounter.getStartTime())/1000f;
    86          System.out.println("Active connections: "+activeConnectionsCounter.getCount());
    87          System.out.println("Echoed bytes: " + (echoedBytesCounter.getCount()/1024f) + " kb");
    88          echoedBytesCounter.reset();
    89      }
    90
    91
    92      public void run() {
    93          try {
    94
    95              activeConnectionsCounter.reset();
    96              echoedBytesCounter.reset();
    97
    98              shutdownLatch = new Latch();
    99
    100             ChannelFactory factory = new ChannelFactory();
    101             AsynchChannelServer server = factory.bindAsynchChannel(url);
    102             System.out.println("Server accepting connections on: "+server.getConnectURI());
    103             server.setAcceptListener(this);
    104             server.start();
    105
    106             while(!shutdownLatch.attempt(sampleInterval)) {
    107                 printSampleData();
    108             }
    109
    110             System.out.println("Stopping server.");
    111             server.stop(1000*5);
    112             server.dispose();
    113
    114         } catch (IOException e) {
    115             e.printStackTrace();
    116         } catch (InterruptedException e) {
    117         }
    118     }
    119
    120     public String getUrl() {
    121         return url.toString();
    122     }
    123
    124     public void setUrl(String url) throws URISyntaxException {
    125         this.url = new URI(url);
    126     }
    127
    128     class ServerConnectionHandler implements AsynchChannelListener {
    129
    130         private final AsynchChannel asynchChannel;
    131         private boolean disposed;
    132
    133         public ServerConnectionHandler(AsynchChannel asynchChannel) {
    134             this.asynchChannel = asynchChannel;
    135             activeConnectionsCounter.increment();
    136         }
    137
    138         public void onPacket(Packet packet) {
    139
    140             if( packet == EOSPacket.EOS_PACKET ) {
    141                 System.out.println("Peer disconnected.");
    142                 dispose();
    143                 return;
    144             }
    145
    146             try {
    147                 if( requestDelay > 0 ) {
    148                     Thread.sleep(requestDelay);
    149                 }
    150
    151                 echoedBytesCounter.add(packet.remaining());
    152                 asynchChannel.write(packet);
    153                 asynchChannel.flush();
    154
    155             } catch (IOException e) {
    156                 onPacketError(e);
    157             } catch (InterruptedException e) {
    158                 System.out.println("Interrupted... Shutting down.");
    159                 dispose();
    160             }
    161         }
    162
    163         public void onPacketError(IOException error) {
    164             error.printStackTrace();
    165             dispose();
    166         }
    167
    168         private void dispose() {
    169             if( !disposed ) {
    170                 asynchChannel.dispose();
    171                 activeConnectionsCounter.decrement();
    172                 disposed=true;
    173             }
    174         }
    175     }
    176
    177     public void onAccept(Channel channel) {
    178         try {
    179
    180             AsynchChannel asynchChannel = SynchToAsynchChannelAdapter.adapt(channel);
    181             asynchChannel.setAsynchChannelListener(new ServerConnectionHandler(asynchChannel));
    182             asynchChannel.start();
    183
    184         } catch (IOException e) {
    185             onAcceptError(e);
    186         }
    187     }
    188
    189     public void onAcceptError(IOException error) {
    190         error.printStackTrace();
    191         shutdownLatch.release();
    192     }
    193
    194     /***
    195      * @return Returns the requestDelay.
    196      */
    197     public long getRequestDelay() {
    198         return requestDelay;
    199     }
    200     /***
    201      * @param requestDelay The requestDelay to set.
    202      */
    203     public void setRequestDelay(long requestDelay) {
    204         this.requestDelay = requestDelay;
    205     }
    206     /***
    207      * @return Returns the sampleInterval.
    208      */
    209     public long getSampleInterval() {
    210         return sampleInterval;
    211     }
    212     /***
    213      * @param sampleInterval The sampleInterval to set.
    214      */
    215     public void setSampleInterval(long sampleInterval) {
    216         this.sampleInterval = sampleInterval;
    217     }
    218 }