Trying out various reading patterns for FileChannel

gistfile1.txt

package voldemort.utils;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.Arrays;
import java.util.Random;

import voldemort.VoldemortException;

public class TestFileChannel {

    public static int NUM_TUPLES = 2000000;
    public static int NUM_ITERS = 2;
    public static int CHECKPOINT = 10000;
    public static long[] operationTimes;
    private static long elapsedTimeNs = 0;
    private static long globalTupleCount = 0;

    public static void Approach1() throws IOException {
        Random r = new Random();
        byte[] dummyKey = new byte[1];
        for(int iter = 0; iter < NUM_ITERS; iter++) {

            System.out.println("Doing iter " + iter);
            File tempFile = new File(createTempDir(), Integer.toString(r.nextInt()));
            System.out.println("Populating data to  " + tempFile.getAbsolutePath());
            writeData(tempFile);
            System.out.println("Finished writing to " + tempFile.getAbsolutePath());
            FileChannel dataFile = openChannel(tempFile);
            operationTimes = new long[(int) globalTupleCount];

            int tupleId = 0;
            long offset = 0;
            long start = System.nanoTime();
            while(tupleId < NUM_TUPLES) {
                if(tupleId % CHECKPOINT == 0) {
                    // System.out.println("Completed reading " + tupleId);
                }
                long start2 = System.nanoTime();
                ByteBuffer numKeyValueSize = ByteBuffer.allocate(ByteUtils.SIZE_OF_BYTE + 2
                                                                 * ByteUtils.SIZE_OF_INT);
                dataFile.read(numKeyValueSize, offset);
                int numKeyValue = numKeyValueSize.get(0) & ByteUtils.MASK_11111111;
                int keySize = numKeyValueSize.getInt(ByteUtils.SIZE_OF_BYTE);
                int valueSize = numKeyValueSize.getInt(ByteUtils.SIZE_OF_BYTE
                                                       + ByteUtils.SIZE_OF_INT);
                offset += ByteUtils.SIZE_OF_BYTE + (2 * ByteUtils.SIZE_OF_INT);

                int keyId = 0;
                do {
                    if(valueSize == -1 && keySize == -1) {
                        start2 = System.nanoTime();
                        numKeyValueSize.clear();
                        dataFile.read(numKeyValueSize, offset);
                        keySize = numKeyValueSize.getInt(0);
                        valueSize = numKeyValueSize.getInt(ByteUtils.SIZE_OF_INT);
                        offset += (2 * ByteUtils.SIZE_OF_INT);
                    }
                    ByteBuffer keyBuffer = ByteBuffer.allocate(keySize);
                    dataFile.read(keyBuffer, offset);
                    offset += keySize;
                    ByteUtils.compare(dummyKey, keyBuffer.array());
                    ByteBuffer valueBuffer = ByteBuffer.allocate(valueSize);
                    dataFile.read(valueBuffer, offset);
                    offset += valueSize;

                    keySize = valueSize = -1;
                    keyId++;
                    operationTimes[tupleId] = System.nanoTime() - start2;
                    tupleId++;
                } while(keyId < numKeyValue);
            }

            elapsedTimeNs = System.nanoTime() - start;
            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 static void Approach2() throws IOException {
        Random r = new Random();
        byte[] dummyKey = new byte[1];
        for(int iter = 0; iter < NUM_ITERS; iter++) {

            System.out.println("Doing iter " + iter);
            File tempFile = new File(createTempDir(), Integer.toString(r.nextInt()));
            System.out.println("Populating data to  " + tempFile.getAbsolutePath());
            writeData(tempFile);
            System.out.println("Finished writing to " + tempFile.getAbsolutePath());
            FileChannel dataFile = openChannel(tempFile);
            operationTimes = new long[(int) globalTupleCount];

            int tupleId = 0;
            long offset = 0;
            long start = System.nanoTime();

            while(tupleId < NUM_TUPLES) {
                if(tupleId % CHECKPOINT == 0) {
                    // System.out.println("Completed reading " + tupleId);
                }
                long start2 = System.nanoTime();
                ByteBuffer numKey = ByteBuffer.allocate(ByteUtils.SIZE_OF_BYTE);
                dataFile.read(numKey, offset);
                int numKeyValue = numKey.get(0) & ByteUtils.MASK_11111111;
                offset += ByteUtils.SIZE_OF_BYTE;

                for(int keyId = 0; keyId < numKeyValue; keyId++) {

                    if(start2 == -1) {
                        start2 = System.nanoTime();
                    }
                    ByteBuffer sizeBuffer = ByteBuffer.allocate(2 * ByteUtils.SIZE_OF_INT);
                    dataFile.read(sizeBuffer, offset);
                    int keySize = sizeBuffer.getInt(0);
                    int valueSize = sizeBuffer.getInt(ByteUtils.SIZE_OF_INT);

                    ByteBuffer keyAndValueBuffer = ByteBuffer.allocate(keySize + valueSize);
                    dataFile.read(keyAndValueBuffer, offset + (2 * ByteUtils.SIZE_OF_INT));

                    ByteUtils.compare(dummyKey, keyAndValueBuffer.array(), 0, keySize);
                    ByteUtils.copy(keyAndValueBuffer.array(), keySize, keySize + valueSize);
                    offset += (keySize + valueSize + (2 * ByteUtils.SIZE_OF_INT));
                    operationTimes[tupleId] = System.nanoTime() - start2;
                    start2 = -1;
                    tupleId++;
                }
            }

            elapsedTimeNs = System.nanoTime() - start;
            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 static void Approach3() throws IOException {
        Random r = new Random();
        byte[] dummyKey = new byte[1];
        for(int iter = 0; iter < NUM_ITERS; iter++) {

            System.out.println("Doing iter " + iter);
            File tempFile = new File(createTempDir(), Integer.toString(r.nextInt()));
            System.out.println("Populating data to  " + tempFile.getAbsolutePath());
            writeData(tempFile);
            System.out.println("Finished writing to " + tempFile.getAbsolutePath());
            FileChannel dataFile = openChannel(tempFile);
            operationTimes = new long[(int) globalTupleCount];

            int tupleId = 0;
            long offset = 0;
            long start = System.nanoTime();
            while(tupleId < NUM_TUPLES) {
                if(tupleId % CHECKPOINT == 0) {
                    // System.out.println("Completed reading " + tupleId);
                }
                long start2 = System.nanoTime();
                ByteBuffer numKey = ByteBuffer.allocate(ByteUtils.SIZE_OF_BYTE);
                dataFile.read(numKey, offset);
                int numKeyValue = numKey.get(0) & ByteUtils.MASK_11111111;
                offset += ByteUtils.SIZE_OF_BYTE;

                for(int keyId = 0; keyId < numKeyValue; keyId++) {

                    if(start2 == -1) {
                        start2 = System.nanoTime();
                    }
                    ByteBuffer sizeBuffer = ByteBuffer.allocate(2 * ByteUtils.SIZE_OF_INT);
                    dataFile.read(sizeBuffer, offset);
                    int keySize = sizeBuffer.getInt(0);
                    int valueSize = sizeBuffer.getInt(ByteUtils.SIZE_OF_INT);
                    offset += (2 * ByteUtils.SIZE_OF_INT);

                    ByteBuffer keyBuffer = ByteBuffer.allocate(keySize);
                    dataFile.read(keyBuffer, offset);
                    offset += keySize;

                    ByteUtils.compare(dummyKey, keyBuffer.array());
                    ByteBuffer valueBuffer = ByteBuffer.allocate(valueSize);
                    dataFile.read(valueBuffer, offset);
                    offset += valueSize;

                    operationTimes[tupleId] = System.nanoTime() - start2;
                    start2 = -1;
                    tupleId++;
                }
            }

            elapsedTimeNs = System.nanoTime() - start;
            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 static void Approach4() throws IOException {
        Random r = new Random();
        byte[] dummyKey = new byte[1];
        for(int iter = 0; iter < NUM_ITERS; iter++) {

            System.out.println("Doing iter " + iter);
            File tempFile = new File(createTempDir(), Integer.toString(r.nextInt()));
            System.out.println("Populating data to  " + tempFile.getAbsolutePath());
            writeData(tempFile);
            System.out.println("Finished writing to " + tempFile.getAbsolutePath());
            FileChannel dataFile = openChannel(tempFile);
            operationTimes = new long[(int) globalTupleCount];

            int tupleId = 0;
            long offset = 0;
            long start = System.nanoTime();
            while(tupleId < NUM_TUPLES) {
                if(tupleId % CHECKPOINT == 0) {
                    // System.out.println("Completed reading " + tupleId);
                }
                long start2 = System.nanoTime();
                ByteBuffer numKeyValueSize = ByteBuffer.allocate(ByteUtils.SIZE_OF_BYTE + 2
                                                                 * ByteUtils.SIZE_OF_INT);
                dataFile.read(numKeyValueSize, offset);
                int numKeyValue = numKeyValueSize.get(0) & ByteUtils.MASK_11111111;
                int keySize = numKeyValueSize.getInt(ByteUtils.SIZE_OF_BYTE);
                int valueSize = numKeyValueSize.getInt(ByteUtils.SIZE_OF_BYTE
                                                       + ByteUtils.SIZE_OF_INT);
                offset += ByteUtils.SIZE_OF_BYTE + (2 * ByteUtils.SIZE_OF_INT);

                int keyId = 0;
                do {
                    if(valueSize == -1 && keySize == -1) {
                        start2 = System.nanoTime();
                        numKeyValueSize.clear();
                        dataFile.read(numKeyValueSize, offset);
                        keySize = numKeyValueSize.getInt(0);
                        valueSize = numKeyValueSize.getInt(ByteUtils.SIZE_OF_INT);
                        offset += (2 * ByteUtils.SIZE_OF_INT);
                    }
                    ByteBuffer keyAndValueBuffer = ByteBuffer.allocate(keySize + valueSize);
                    dataFile.read(keyAndValueBuffer, offset);
                    offset += (keySize + valueSize);

                    ByteUtils.compare(dummyKey, keyAndValueBuffer.array(), 0, keySize);
                    ByteUtils.copy(keyAndValueBuffer.array(), keySize, keySize + valueSize);

                    keySize = valueSize = -1;

                    keyId++;
                    operationTimes[tupleId] = System.nanoTime() - start2;
                    tupleId++;
                } while(keyId < numKeyValue);
            }

            elapsedTimeNs = System.nanoTime() - start;
            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 static void main(String args[]) throws IOException {
        System.out.println("======== APPROACH 4 =======");
        Approach4();
        System.out.println("======== APPROACH 3 =======");
        Approach3();
        System.out.println("======== APPROACH 2 =======");
        Approach2();
        System.out.println("======== APPROACH 1 =======");
        Approach1();

    }

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

    public static double getOperationTimeMsQuantile(double quantile) {
        return quantile(operationTimes, quantile) / (double) Time.NS_PER_MS;
    }

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

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

    public static void writeData(File tempData) throws IOException {
        FileOutputStream stream = new FileOutputStream(tempData);
        Random r = new Random();
        int tupleId = 0;
        while(tupleId < NUM_TUPLES) {

            float f = r.nextFloat();
            int numKeys = 1;
            if(f > 0.90) {
                numKeys = 2;
            }
            stream.write(numKeys);
            for(int numKey = 0; numKey < numKeys; numKey++) {
                int keySize = r.nextInt(500);
                int valueSize = r.nextInt(4096);
                byte[] key = randomBytes(keySize);
                byte[] value = randomBytes(valueSize);

                byte[] sizeArray = new byte[ByteUtils.SIZE_OF_INT];
                ByteUtils.writeInt(sizeArray, keySize, 0);
                stream.write(sizeArray);

                ByteUtils.writeInt(sizeArray, valueSize, 0);
                stream.write(sizeArray);
                stream.write(key);
                stream.write(value);
                tupleId++;
            }
        }
        globalTupleCount = tupleId;
        System.out.println("Last tuple id written - " + tupleId);
        stream.flush();
        stream.close();
    }

    private static FileChannel openChannel(File file) {
        try {
            return new FileInputStream(file).getChannel();
        } catch(IOException e) {
            throw new VoldemortException(e);
        }
    }

    public static File createTempDir() {
        File parent = new File(System.getProperty("java.io.tmpdir"));
        File temp = new File(parent, Integer.toString(Math.abs(new Random().nextInt()) % 1000000));
        temp.delete();
        temp.mkdir();
        temp.deleteOnExit();
        return temp;
    }

    /**
     * Generate an array of random bytes
     * 
     * @param length
     * @return
     */
    public static byte[] randomBytes(int length) {
        byte[] bytes = new byte[length];
        new Random().nextBytes(bytes);
        return bytes;
    }

    /**
     * Compute the mean of the given values
     * 
     * @param values The values
     * @return The mean
     */
    public static double mean(long[] values) {
        double total = 0.0;
        for(int i = 0; i < values.length; i++)
            total += values[i];
        return total / values.length;
    }

    public static long quantile(long[] values, double quantile) {
        if(values == null)
            throw new IllegalArgumentException("Values cannot be null.");
        if(quantile < 0.0 || quantile > 1.0)
            throw new IllegalArgumentException("Quantile must be between 0.0 and 1.0");

        long[] copy = new long[values.length];
        System.arraycopy(values, 0, copy, 0, copy.length);
        Arrays.sort(copy);
        int index = (int) (copy.length * quantile);
        return copy[index];
    }
}