PerformanceTest

/*
 * Copyright 2008-2009 LinkedIn, Inc
 * 
 * Licensed 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 voldemort.performance;

import java.util.Arrays;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import voldemort.TestUtils;
import voldemort.utils.Pair;
import voldemort.utils.Time;

public abstract class PerformanceTest {

    private final AtomicInteger numberOfFailures = new AtomicInteger(0);
    private long elapsedTimeNs;
    private long[] operationTimes;
    private volatile boolean hasCompleted;
    private int numberOfThreads;
    private final AtomicInteger index = new AtomicInteger(0);

    public abstract void doOperation(int index) throws Exception;

    public void setUp() {
    // override me to do stuff
    }

    public void tearDown() {
    // override me to do stuff
    }

    public void run(int numRequests, int numThreads) {
        setUp();
        try {
            this.numberOfThreads = numThreads;
            this.hasCompleted = false;
            this.numberOfFailures.set(0);
            this.operationTimes = new long[numRequests];
            ExecutorService executor = Executors.newFixedThreadPool(numThreads);
            final CountDownLatch latch = new CountDownLatch(numRequests);

            long start = System.nanoTime();
            for(int i = 0; i < numRequests; i++) {
                executor.execute(new Runnable() {

                    public void run() {
                        int current = index.getAndIncrement();
                        long begin = System.nanoTime();
                        try {
                            doOperation(current);
                        } catch(Exception e) {
                            numberOfFailures.getAndIncrement();
                            e.printStackTrace();
                        } finally {
                            operationTimes[current] = System.nanoTime() - begin;
                            latch.countDown();
                        }
                    }
                });
            }

            try {
                latch.await();
            } catch(InterruptedException e) {
                e.printStackTrace();
            }

            this.hasCompleted = true;
            this.elapsedTimeNs = System.nanoTime() - start;
            executor.shutdownNow();
            try {
                executor.awaitTermination(3, TimeUnit.SECONDS);
            } catch(InterruptedException e) {}
        } finally {
            tearDown();
        }
    }

    public void printStats() {
        checkComplete();
        System.out.println("Total number of operations: " + this.operationTimes.length);
        System.out.println("Total elapsed seconds: " + this.elapsedTimeNs
                           / (double) Time.NS_PER_SECOND);
        System.out.println("Number of failures: " + this.numberOfFailures.get());
        System.out.println("Number of threads: " + this.numberOfThreads);
        System.out.println("Avg. operations/second: " + getOperationsPerSecond());
        System.out.println("Average time: " + getAverageOperationTimeMs() + " ms");
        System.out.println("Std dev.: " + getStandardDeviationMs() + " ms");
        System.out.println("Median time: " + getOperationTimeMsQuantile(0.5d) + " ms");
        System.out.println("1st percentile: " + getOperationTimeMsQuantile(0.01d) + " ms");
        System.out.println("99th percentile: " + getOperationTimeMsQuantile(0.99d) + " ms");
    }

    public double getOperationsPerSecond() {
        // checkComplete();
        double elapsedSeconds = this.elapsedTimeNs / (double) Time.NS_PER_SECOND;
        return this.operationTimes.length / elapsedSeconds;
    }

    public Pair<Double, Double> getMediaAnd99() {
        long[] tempResults = new long[index.get()];
        System.arraycopy(this.operationTimes, 0, tempResults, 0, tempResults.length);

        Arrays.sort(tempResults);
        int medianIndex = (int) (tempResults.length * 0.5);
        int index99 = (int) (tempResults.length * 0.99);

        return Pair.create(tempResults[medianIndex] / (double) Time.NS_PER_MS,
                           tempResults[index99] / (double) Time.NS_PER_MS);
    }

    public double getOperationTimeMsQuantile(double quantile) {
        // checkComplete();
        long[] tempResults = new long[index.get()];
        System.arraycopy(this.operationTimes, 0, tempResults, 0, tempResults.length);
        return TestUtils.quantile(tempResults, quantile) / (double) Time.NS_PER_MS;
    }

    public double getAverageOperationTimeMs() {
        // checkComplete();
        double mean = TestUtils.mean(this.operationTimes);
        return mean / Time.NS_PER_MS;
    }

    public double getStandardDeviationMs() {
        // checkComplete();
        double mean = TestUtils.mean(this.operationTimes);
        double sum = 0.0;
        for(int i = 0; i < this.operationTimes.length; i++)
            sum += (this.operationTimes[i] - mean) * (this.operationTimes[i] - mean);
        return Math.sqrt(sum / this.operationTimes.length) / Time.NS_PER_MS;
    }

    private void checkComplete() {
        if(!hasCompleted)
            throw new RuntimeException("Hasn't finished running yet!");
    }

}

ReadOnlyStoreSwitchPerformance

/*
 * Copyright 2008-2009 LinkedIn, Inc
 * 
 * Licensed 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 voldemort.performance;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

import joptsimple.OptionParser;
import joptsimple.OptionSet;
import voldemort.client.ClientConfig;
import voldemort.client.SocketStoreClientFactory;
import voldemort.client.StoreClient;
import voldemort.cluster.Cluster;
import voldemort.cluster.Node;
import voldemort.routing.RoutingStrategy;
import voldemort.routing.RoutingStrategyFactory;
import voldemort.serialization.Serializer;
import voldemort.serialization.json.JsonTypeDefinition;
import voldemort.serialization.json.JsonTypeSerializer;
import voldemort.store.StoreDefinition;
import voldemort.utils.CmdUtils;
import voldemort.utils.Pair;
import voldemort.versioning.Versioned;
import voldemort.xml.ClusterMapper;
import voldemort.xml.StoreDefinitionsMapper;

public class ReadOnlyStoreSwitchPerformance {

    public static void main(String[] args) throws FileNotFoundException, IOException {
        OptionParser parser = new OptionParser();
        parser.accepts("help", "print usage information");
        parser.accepts("url", "[REQUIRED] url").withRequiredArg().ofType(String.class);
        parser.accepts("requests", "number of requests ( default : 100000 ) ")
              .withRequiredArg()
              .ofType(Integer.class);
        parser.accepts("cluster-xml", "[REQUIRED] Path to cluster.xml")
              .withRequiredArg()
              .describedAs("path");
        parser.accepts("store-xml", "[REQUIRED] Path to stores.xml")
              .withRequiredArg()
              .describedAs("path");
        parser.accepts("store-name", "[REQUIRED] Store name")
              .withRequiredArg()
              .describedAs("store-name");

        OptionSet options = parser.parse(args);

        if(options.has("help")) {
            parser.printHelpOn(System.out);
            System.exit(0);
        }

        CmdUtils.croakIfMissing(parser, options, "url", "requests", "cluster-xml", "store-xml");

        final String url = (String) options.valueOf("url");
        final int numRequests = CmdUtils.valueOf(options, "requests", 100000);
        final String clusterXmlPath = (String) options.valueOf("cluster-xml");
        final String storeXmlPath = (String) options.valueOf("store-xml");
        final String storeName = (String) options.valueOf("store-name");

        final Cluster cluster = new ClusterMapper().readCluster(new File(clusterXmlPath));
        final List<StoreDefinition> storeDefs = new StoreDefinitionsMapper().readStoreList(new File(storeXmlPath));
        StoreDefinition storeDef = null;
        for(StoreDefinition currentStoreDef: storeDefs) {
            if(currentStoreDef.getName().compareTo(storeName) == 0) {
                storeDef = currentStoreDef;
            }
        }
        if(storeDef == null) {
            System.err.println("Could not find store " + storeName);
            System.exit(-1);
        }

        final AtomicInteger unreachableResults = new AtomicInteger(0);
        final AtomicInteger nullResults = new AtomicInteger(0);
        final AtomicInteger totalResults = new AtomicInteger(0);

        final BlockingQueue<Integer> requestIds = new ArrayBlockingQueue<Integer>(20000);
        final Executor executor = Executors.newFixedThreadPool(1);

        final Serializer<Object> keySerializer = new JsonTypeSerializer(JsonTypeDefinition.fromJson("'int32'"),
                                                                        true);
        final RoutingStrategy strategy = new RoutingStrategyFactory().updateRoutingStrategy(storeDef,
                                                                                            cluster);

        Runnable requestGenerator = new Runnable() {

            public void run() {
                System.out.println("Generating random requests.");
                Random random = new Random();

                while(true) {
                    try {
                        // Add keys only for which the
                        int request = random.nextInt(numRequests);
                        List<Node> nn = strategy.routeRequest(keySerializer.toBytes(request));
                        if(nn.get(0).getId() == 0 && nn.get(1).getId() == 1) {
                            requestIds.put(request);
                        }
                    } catch(InterruptedException e) {
                        e.printStackTrace();
                    }
                }

            }
        };
        executor.execute(requestGenerator);

        final AtomicInteger current = new AtomicInteger();
        SocketStoreClientFactory factory = new SocketStoreClientFactory(new ClientConfig().setBootstrapUrls(url));
        final StoreClient<Integer, Object> store = factory.getStoreClient(storeName);

        final PerformanceTest readWriteTest = new PerformanceTest() {

            @Override
            public void doOperation(int index) throws Exception {
                try {
                    totalResults.incrementAndGet();
                    int curr = current.getAndIncrement();
                    Versioned<Object> results = store.get(requestIds.take(), null);
                    if(curr % 5000 == 0) {
                        Pair<Double, Double> result = getMediaAnd99();
                        System.out.println(curr + "\t" + result.getFirst() + "\t"
                                           + result.getSecond());
                    }

                    if(results == null || results.getValue() == null)
                        nullResults.incrementAndGet();

                } catch(Exception e) {
                    unreachableResults.incrementAndGet();
                }
            }
        };
        System.out.println("Running test...");
        readWriteTest.run(numRequests, 10);
        System.out.println("Random Access Read Only store Results:");
        System.out.println("Null reads ratio:" + (nullResults.doubleValue())
                           / totalResults.doubleValue());
        System.out.println("Unreachable exceptions - " + unreachableResults.get());
        readWriteTest.printStats();
        System.exit(0);
    }
}