    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.ArrayList;
    24  import java.util.HashMap;
    25  import java.util.Iterator;
    26
    27  import org.activeio.ChannelFactory;
    28  import org.activeio.Packet;
    29  import org.activeio.SynchChannel;
    30  import org.activeio.packet.ByteArrayPacket;
    31  import org.activeio.packet.EOSPacket;
    32  import org.activeio.stats.CountStatisticImpl;
    33  import org.activeio.stats.TimeStatisticImpl;
    34  import org.apache.commons.beanutils.BeanUtils;
    35
    36  import EDU.oswego.cs.dl.util.concurrent.Latch;
    37  import EDU.oswego.cs.dl.util.concurrent.PooledExecutor;
    38
    39  /***
    40   * Simulates multiple a simple tcp echo clients for use in benchmarking activeio
    41   * channel implementations.
    42   *
    43   * @version $Revision$
    44   */
    45  public class ClientLoadSimulator implements Runnable {
    46
    47      private URI url;
    48
    49      // Afects how clients are created
    50      private int concurrentClients = 10;
    51      private long rampUpTime = 1000 * concurrentClients;
    52
    53      // Afects how clients behave
    54      private long requestDelay = 500;
    55      private int requestIterations = Integer.MAX_VALUE;
    56      private int requestSize = 1024;
    57
    58      PooledExecutor threadPool = new PooledExecutor();
    59
    60      // The packet the clients send to the server.
    61      private Packet requestPacket;
    62      private long sampleInterval = 1000;
    63      private ChannelFactory factory = new ChannelFactory();
    64      private Latch shutdownLatch;
    65
    66      private final CountStatisticImpl activeConnectionsCounter = new CountStatisticImpl("activeConnectionsCounter",
    67              "The number of active connection attached to the server.");
    68      private final CountStatisticImpl echoedBytesCounter = new CountStatisticImpl("echoedBytesCounter",
    69              "The number of bytes that have been echoed by the server.");
    70      private final TimeStatisticImpl requestLatency = new TimeStatisticImpl("requestLatency",
    71              "The amount of time that is spent waiting for a request to be serviced");
    72
    73      public static void main(String[] args) throws URISyntaxException, IllegalAccessException, InvocationTargetException {
    74
    75          ClientLoadSimulator client = new ClientLoadSimulator();
    76
    77          HashMap options = new HashMap();
    78          for (int i = 0; i < args.length; i++) {
    79
    80              String option = args[i];
    81              if (!option.startsWith("-") || option.length() < 2 || i + 1 >= args.length) {
    82                  System.out.println("Invalid usage");
    83                  return;
    84              }
    85
    86              option = option.substring(1);
    87              options.put(option, args[++i]);
    88          }
    89
    90          BeanUtils.populate(client, options);
    91
    92          System.out.println();
    93          System.out.println("Server starting with the following options: ");
    94          System.out.println(" url="+client.getUrl());
    95          System.out.println(" sampleInterval="+client.getSampleInterval());
    96          System.out.println(" concurrentClients="+client.getConcurrentClients());
    97          System.out.println(" rampUpTime="+client.getRampUpTime());
    98          System.out.println(" requestIterations="+client.getRequestIterations());
    99          System.out.println(" requestSize="+client.getRequestSize());
    100         System.out.println(" requestDelay="+client.getRequestDelay());
    101         System.out.println();
    102         client.run();
    103
    104     }
    105     private void printSampleData() {
    106         long now = System.currentTimeMillis();
    107         float runDuration = (now - activeConnectionsCounter.getStartTime()) / 1000f;
    108         System.out.println("Active connections: " + activeConnectionsCounter.getCount());
    109         System.out.println("Echoed bytes: " + (echoedBytesCounter.getCount()/1024f) + " kb"
    110                 + ", Request latency: " + requestLatency.getAverageTime()+" ms");
    111         echoedBytesCounter.reset();
    112         requestLatency.reset();
    113     }
    114
    115     public void run() {
    116         ArrayList clients = new ArrayList();
    117         try {
    118
    119             shutdownLatch = new Latch();
    120             activeConnectionsCounter.reset();
    121             echoedBytesCounter.reset();
    122
    123             new Thread("Sampler") {
    124                 public void run() {
    125                     System.out.println("Sampler started.");
    126                     try {
    127                         while (!shutdownLatch.attempt(sampleInterval)) {
    128                             printSampleData();
    129                         }
    130                     } catch (InterruptedException e) {
    131                     }
    132                     System.out.println("Sampler stopped.");
    133                 }
    134             }.start();
    135
    136             byte data[] = new byte[requestSize];
    137             for (int i = 0; i < data.length; i++) {
    138                 data[i] = (byte) i;
    139             }
    140             requestPacket = new ByteArrayPacket(data);
    141
    142             // Loop to ramp up the clients.
    143
    144             long clientActivationDelay = rampUpTime / concurrentClients;
    145             for (int i = 0; i < concurrentClients && !shutdownLatch.attempt(clientActivationDelay); i++) {
    146                 System.out.println("Adding Client: " + i);
    147                 Client client = new Client();
    148                 clients.add(client);
    149                 new Thread(client, "Client: " + i).start();
    150             }
    151
    152             shutdownLatch.acquire();
    153
    154         } catch (InterruptedException e) {
    155         } finally {
    156             System.out.println("Shutting down clients.");
    157             for (Iterator iter = clients.iterator(); iter.hasNext();) {
    158                 Client client = (Client) iter.next();
    159                 client.dispose();
    160             }
    161         }
    162     }
    163
    164     public String getUrl() {
    165         return url.toString();
    166     }
    167
    168     public void setUrl(String url) throws URISyntaxException {
    169         this.url = new URI(url);
    170     }
    171
    172     class Client implements Runnable {
    173
    174         private Latch shutdownLatch = new Latch();
    175
    176         Packet packet = requestPacket.duplicate();
    177
    178         private SynchChannel synchChannel;
    179
    180         public void run() {
    181             try {
    182                 System.out.println("Client started.");
    183
    184                 activeConnectionsCounter.increment();
    185                 synchChannel = factory.openSynchChannel(url);
    186                 for (int i = 0; i < requestIterations && !shutdownLatch.attempt(1) ; i++) {
    187
    188                     long start = System.currentTimeMillis();
    189                     sendRequest();
    190                     long end = System.currentTimeMillis();
    191
    192                     requestLatency.addTime(end - start);
    193                     echoedBytesCounter.add(packet.remaining());
    194
    195                     if( requestDelay > 0 ) {
    196                         Thread.sleep(requestDelay);
    197                     }
    198                 }
    199
    200             } catch (IOException e) {
    201                 e.printStackTrace();
    202             } catch (InterruptedException e) {
    203                 e.printStackTrace();
    204             } finally {
    205                 System.out.println("Client stopped.");
    206
    207                 activeConnectionsCounter.decrement();
    208                 if( synchChannel!=null ) {
    209                     synchChannel.dispose();
    210                     synchChannel = null;
    211                 }
    212             }
    213         }
    214
    215         private void sendRequest() throws IOException, InterruptedException {
    216
    217             final Latch done = new Latch();
    218
    219             // Read the data async to avoid dead locks due buffers being to small for
    220             // data being sent.
    221             threadPool.execute(new Runnable() {
    222                 public void run() {
    223                     try {
    224                         int c = 0;
    225                         while (c < packet.remaining()) {
    226                             Packet p = synchChannel.read(1000*5);
    227                             if( p==null ) {
    228                                 continue;
    229                             }
    230                             if( p == EOSPacket.EOS_PACKET ) {
    231                                 System.out.println("Peer disconnected.");
    232                                 dispose();
    233                             }
    234                             c += p.remaining();
    235                         }
    236                         done.release();
    237                     } catch (IOException e) {
    238                         e.printStackTrace();
    239                     }
    240                 }
    241             });
    242
    243             synchChannel.write(packet.duplicate());
    244             synchChannel.flush();
    245             done.acquire();
    246
    247         }
    248
    249         public void dispose() {
    250             shutdownLatch.release();
    251         }
    252     }
    253
    254     /***
    255      * @return Returns the concurrentClients.
    256      */
    257     public int getConcurrentClients() {
    258         return concurrentClients;
    259     }
    260
    261     /***
    262      * @param concurrentClients
    263      *            The concurrentClients to set.
    264      */
    265     public void setConcurrentClients(int concurrentClients) {
    266         this.concurrentClients = concurrentClients;
    267     }
    268
    269     /***
    270      * @return Returns the rampUpTime.
    271      */
    272     public long getRampUpTime() {
    273         return rampUpTime;
    274     }
    275
    276     /***
    277      * @param rampUpTime
    278      *            The rampUpTime to set.
    279      */
    280     public void setRampUpTime(long rampUpTime) {
    281         this.rampUpTime = rampUpTime;
    282     }
    283
    284     /***
    285      * @return Returns the requestDelay.
    286      */
    287     public long getRequestDelay() {
    288         return requestDelay;
    289     }
    290
    291     /***
    292      * @param requestDelay
    293      *            The requestDelay to set.
    294      */
    295     public void setRequestDelay(long requestDelay) {
    296         this.requestDelay = requestDelay;
    297     }
    298
    299     /***
    300      * @return Returns the requestIterations.
    301      */
    302     public int getRequestIterations() {
    303         return requestIterations;
    304     }
    305
    306     /***
    307      * @param requestIterations
    308      *            The requestIterations to set.
    309      */
    310     public void setRequestIterations(int requestIterations) {
    311         this.requestIterations = requestIterations;
    312     }
    313
    314     /***
    315      * @return Returns the requestSize.
    316      */
    317     public int getRequestSize() {
    318         return requestSize;
    319     }
    320
    321     /***
    322      * @param requestSize
    323      *            The requestSize to set.
    324      */
    325     public void setRequestSize(int requestSize) {
    326         this.requestSize = requestSize;
    327     }
    328
    329     /***
    330      * @return Returns the sampleInterval.
    331      */
    332     public long getSampleInterval() {
    333         return sampleInterval;
    334     }
    335     /***
    336      * @param sampleInterval The sampleInterval to set.
    337      */
    338     public void setSampleInterval(long sampleInterval) {
    339         this.sampleInterval = sampleInterval;
    340     }
    341 }