continousthroughputtest.java

来自「JGRoups源码」· Java 代码 · 共 462 行 · 第 1/2 页

            System.out.println(ex);
        }
    }

    /**
     * sendBurst(int n): sends a burst of messages with small payload
     */

    void sendBurst(long n) {
        try {
            byte[] buf=Util.objectToByteBuffer("Standard Mex");
            for(int i=0; i < n; i++) {
                channel.send(new Message(null, null, buf));
            }
        }
        catch(Exception ex) {
            System.out.println("sendBurst: " + ex);
        }
    }


    /**
     * showStats: Prints resulting times and troughput
     */

    void showStats(long start, long stop, long messages, int burstlength) {
        String result;
        long elapsedTime=(stop - start);
        long troughPut=(messages * 1000) / elapsedTime;
//    troughputSum += troughPut;
        maxTroughput=(maxTroughput > troughPut) ? maxTroughput : troughPut;
        minTroughput=(minTroughput < troughPut) ? minTroughput : troughPut;
//    System.out.println("Elapsed Time: " + (stop-start) + " milliseconds to receive " + messages + " messages");
        result="Elapsed Time: " + (stop - start) +
                "| messages:" + messages +
                "| burst length:" + burstlength +
                "| Troughput:" + troughPut +
                "| max: " + maxTroughput +
                "| min: " + minTroughput +
                "\r\n";
        System.out.println(result);
        try {
            logWriter.write(result);
            logWriter.flush();
        }
        catch(Exception ex) {
            System.out.println("showStats():" + ex);
        }

    }

    int askNumber(BufferedReader reader, String text) {
        int number=0;
        String input="10";
        System.out.println(text);
        try {
            input=reader.readLine();
        }
        catch(Exception ex) {
            System.out.println("AskNumber :" + ex);
        }

        number=Integer.parseInt(input);
        return number;
    }

    /**
     * Resets stored statistics and counters
     */

    void resetData() {
        maxTroughput=0;
        minTroughput=10000;
        meanTroughput=0;
        numTests=0;
        troughputSum=0;
    }

    /**
     * Make a pingpong test:
     * For n times a message is sent and view.size() messages are received
     * Every 1000 messages sent the throughput is evaluated or at the end of the test
     */
    void pingpongTest(long n, int burst_length, boolean partialResultsPrint) {
        long i=0;
        long start=System.currentTimeMillis();
        long tempstart=System.currentTimeMillis();
        long stop, throughput;
        try {
            for(i=0; i < n; i++) {
                for(int k=0; k < burst_length; k++)
                    channel.send(new Message(null, null, new Long(i)));
                for(int j=0; j < (view.size() * burst_length); j++) {
                    channel.receive(20000);
                }
                if(partialResultsPrint && ((i % 1000) == 0)) {
                    if(i == 0) continue;
                    stop=System.currentTimeMillis();
                    throughput=(1000000 / (stop - tempstart)) * view.size() * burst_length;
                    try {
                        System.out.println(new Date(stop).toString() + " : " + throughput);
                        logWriter.write(new Date(stop).toString() + " : " + throughput);
                        logWriter.write("\r\n");
                        logWriter.flush();
                        tempstart=System.currentTimeMillis();
                    }
                    catch(Exception ex) {
                        ex.printStackTrace();
                    }
                }
            }
        }
        catch(TimeoutException ex) {
            System.out.println("Timeout Receiving, round: " + i);
            System.exit(5);
        }
        catch(Exception ex) {
            ex.printStackTrace();
            System.exit(4);
        }
        stop=System.currentTimeMillis();
        showStats(start, stop, n * view.size() * burst_length, burst_length);
    }

    void sweep(long tests, int burstlenght) {
        long messagespertest=10000;
        for(int i=0; i < tests; i++) {
            burstlenght+=i;
            pingpongTest(messagespertest / burstlenght, burstlenght, false);
        }
    }

    /**
     * Makes a continous test handling view changes
     */
    void cpingTest(int burst_lenght, boolean printoutput) {
        Object recvd=null;
        long start=System.currentTimeMillis();
        for(long i=1; i < Long.MAX_VALUE; i++) {
//	 System.out.println("Round: " + i);
            try {
                channel.send(null, null, "cping");
                for(int j=0; j < burst_lenght * view.size();) {
                    recvd=channel.receive(10000);
                    if(recvd instanceof View) {
                        view=(View)recvd;
                        System.out.println(view);
                        log(view.toString());
                    }
                    else {
                        j++;
                    }
                }
            }
            catch(TimeoutException tex) {
                try {
                    channel.send(new Message(null, null, "cping"));
                    System.out.println("Resent a message for timeout");
                    log("Resent a message for timeout");
                }
                catch(Exception ex) {
                    System.exit(9);
                }
            }
            catch(Exception ex) {
                System.exit(9);
            }
            if((i % 1000) == 0) {
                long stop=System.currentTimeMillis();
                long throughput=i * 1000 * view.size() / (stop - start);
                System.out.println("Througputh = " + throughput);
                log("Througputh = " + throughput);
                start=System.currentTimeMillis();
                i=0;
            }
        }
    }

    /**
     * Used to print messages lenght and their serialized contents.
     */

    public static class MessageLenghtObserver implements ProtocolObserver {

        public void setProtocol(Protocol prot) {
            /** todo: Implement this org.jgroups.debug.ProtocolObserver method*/
            throw new java.lang.UnsupportedOperationException("Method setProtocol() not yet implemented.");
        }

        public boolean up(Event evt, int num_evts) {
            /** todo: Implement this org.jgroups.debug.ProtocolObserver method*/
            throw new java.lang.UnsupportedOperationException("Method up() not yet implemented.");
        }

        public boolean passUp(Event evt) {
            return true;
        }

        public boolean down(Event evt, int num_evts) {
            return true;
        }

        public boolean passDown(Event evt) {
            byte[] buf=null;
            if(evt.getType() == Event.MSG)
                try {
                    buf=Util.objectToByteBuffer(evt.getArg());
                    System.out.println("UDP: sending a message of " +
                            buf.length +
                            "bytes");
                    System.out.println("Message was :");
                    System.out.println(new String(buf));
                }
                catch(Exception ex) {

                }
            return true;
        }
    }

    void log(String str) {
        try {
            logWriter.write(str + "\r\n");
            logWriter.flush();
        }
        catch(Exception ex) {

        }
    }
}