// Copyright (C) 2014-2016 Roger L. Deran. All Rights Reserved.
//
//  May 20, 2016        Roger L. Deran
//
// THIS SOFTWARE CONTAINS CONFIDENTIAL INFORMATION AND TRADE SECRETS
// OF Roger L Deran.  USE, DISCLOSURE, OR REPRODUCTION IS PROHIBITED
// WITHOUT THE PRIOR EXPRESS WRITTEN PERMISSION OF Roger L Deran.
//
// Roger L Deran. MAKES NO REPRESENTATIONS OR WARRANTIES ABOUT
// THE SUITABILITY OF THE SOFTWARE, EITHER EXPRESS OR IMPLIED,
// INCLUDING BUT NOT LIMITED TO THE IMPLIED WARRANTIES OF
// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE, OR
// NON-INFRINGEMENT. Roger L Deran. SHALL NOT BE LIABLE FOR
// ANY DAMAGES SUFFERED BY LICENSEE AS A RESULT OF USING,
// MODIFYING OR DISTRIBUTING THIS SOFTWARE OR ITS DERIVATIVES.

package com.infinitydb.map;

import java.security.SecureRandom;
import java.util.ArrayList;
import java.util.Map;
import java.util.NavigableMap;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.BiConsumer;
import java.util.stream.Collectors;

import com.infinitydb.map.air.AirConcurrentMap;
import com.infinitydb.map.lambda.CollectingThreadedMapVisitor;
import com.infinitydb.map.lambda.MapVisitorBiConsumerWrapper;
import com.infinitydb.map.lambda.ValueCollectingThreadedMapVisitor;
import com.infinitydb.map.visitor.ThreadedMapVisitor;
import com.infinitydb.map.visitor.VisitableMap;

/**
 * @author Roger
 *
 */

public class TwitterDemo {
    public static void main(String... args) {
        try {
//            args = new String[] { "100", "100" };
            TwitterDemoForMapPut.main(args);
            TwitterDemoForMapMemorySize.main(args);
            TwitterDemoForMapIterator.main(args);
            TwitterDemoForNavigableMapGet.main(args);
            TwitterDemoForThreadedGet.main(args);
            TwitterDemoForThreadedLambda.main(args);
            TwitterDemoForBiConsumerWithLambdaInteger.main(args);
            TwitterDemoForThreadedSummerPerformance.main(args);
            TwitterDemoForStreamsCollectorsVsAirConcurrentMap.main(args);
            TwitterDemoForStreamsCollectorsVsAirConcurrentMapArrayList.main(args);
        } catch (Exception e) {
            e.printStackTrace();
            throw e;
        }
    }
}

/**
 * How to determine a Map's put speed simply. Both are ConcurrentNavigableMaps,
 * which are ordered, for fairness. ACM reaches 600K/s and CSLM reaches 351K/s
 */
class TwitterDemoForMapPut {
    public static void main(String... args) {
        System.out.println("TwitterDemoForMapPut");
        final int MAX_SIZE = args.length >= 1 ? Integer.valueOf(args[0]) : 20_000_000;
        final int STEP_SIZE = args.length >= 2 ? Integer.valueOf(args[1]) : 1000_000;
        Random random = new Random();
        Map<Long, Long> map = new AirConcurrentMap<Long, Long>();
        // Map<Long, Long> map = new ConcurrentSkipListMap<Long, Long>();
        // Map<Long, Long> map = new ConcurrentHashMap<Long, Long>();
        // Map<Long, Long> map = new HashMap<Long, Long>();
        // Map<Long, Long> map = new TreeMap<Long, Long>();
        long t0 = System.nanoTime();
        for (int i = 0; i < MAX_SIZE;) {
            for (int j = 0; j < STEP_SIZE; j++) {
                Long n = random.nextLong();
                map.put(n, n);
                i++;
            }
            long t1 = System.nanoTime();
            double entriesPerSecond = i * 1e9 / (t1 - t0);
            System.out.println("Entries=" + i + " Entries/s=" +
                    entriesPerSecond / 1e6 + "M");
        }
    }
}

/**
 * Simple code to show the performance of get() and higher() of
 * ConcurrentNavigableMaps over various sizes.
 * 
 * <pre>
 * AirConcurrentMap:      Entries=10000000 gets/s=817.472K
 * ConcurrentSkipListMap: Entries=10000000 gets/s=390.339K
 * TreeMap:               Entries=10000000 gets/s=919.522K
 * </pre>
 */
class TwitterDemoForNavigableMapGet {
    static final int ITERATIONS = 100_000;

    public static void main(String... args) {
        System.out.println("TwitterDemoForNavigableMapGet");
        final int MAX_SIZE = args.length >= 1 ? Integer.valueOf(args[0]) : 10_000_000;
        final int STEP_SIZE = args.length >= 2 ? Integer.valueOf(args[1]) : 1000_000;
        Random random = new Random();
        ConcurrentNavigableMap<Integer, Integer> map = new
                AirConcurrentMap<Integer, Integer>();
        // ConcrentNavigableMap<Integer, Integer> map = new
        // ConcurrentSkipListMap<Integer, Integer>();
        // NavigableMap<Integer, Integer> map = new
        // TreeMap<Integer, Integer>();
        // Without a tangible result, the loop is optimized out!
        long total = 0;
        for (int i = 0; i < MAX_SIZE;) {
            for (int j = 0; j < STEP_SIZE; j++) {
                Integer n = new Integer(random.nextInt());
                map.put(n, n);
                i++;
            }
            long t0 = System.nanoTime();
            for (int k = 0; k < ITERATIONS; k++) {
                Integer n = new Integer(random.nextInt());
                // map.get(n);
                // higherKey et al are only in NavigableMaps, which are ordered
                map.higherKey(n);
            }
            long t1 = System.nanoTime();
            double seconds = (t1 - t0) / 1e9;
            System.out.printf("Entries=%d gets/s=%6.3fK\n", i,
                    ITERATIONS / seconds / 1e3);
        }
        System.out.println("total=" + total);
    }
}

/**
 * A simple threading test for get() for ConcurrentNavigableMaps.
 * AirConcurrentMap wins.
 * 
 * <pre>
 * AirConcurrentMap:      Entries=10000000 gets/s=549.281K. 
 * ConcurrentSkipListMap: Entries=10000000 gets/s=274.786K.
 * </pre>
 */
class TwitterDemoForThreadedGet {
    static final int THREADS = 8;
    static final int ITERATIONS = 3_000_000;

    static final NavigableMap<Integer, Integer> map = new
            AirConcurrentMap<Integer, Integer>();
    // static final NavigableMap<Integer, Integer> map =
    // new ConcurrentSkipListMap<Integer, Integer>();

    static int entryCount = 0;
    // increment this if the get() is commented out to avoid the loop being
    // optimized away.
    static volatile int doNothing;

    public static void main(String... args) {
        System.out.println("TwitterDemoForThreadedGet");
        final int MAX_SIZE = args.length >= 1 ? Integer.valueOf(args[0]) : 10_000_000;
        final int STEP_SIZE = args.length >= 2 ? Integer.valueOf(args[1]) : 1000_000;
        Random random = new Random(new SecureRandom().nextInt());
        while (entryCount < MAX_SIZE) {
            for (int i = 0; i < STEP_SIZE; i++) {
                Integer n = new Integer(random.nextInt());
                map.put(n, n);
                entryCount++;
            }
            Thread[] threads = new Thread[THREADS];
            for (int j = 0; j < THREADS; j++)
                threads[j] = new Thread() {
                    public void run() {
                        Random random = new Random(new SecureRandom().nextInt());
                        long t0 = System.nanoTime();
                        for (int k = 0; k < ITERATIONS; k++) {
                            Integer n = new Integer(random.nextInt());
                            // map.get(n);
                            map.higherKey(n);
                            // doNothing++; Watch for being optimized out
                            // though!
                        }
                        long t1 = System.nanoTime();
                        double seconds = (t1 - t0) / 1e9;
                        System.out.printf("Entries=%d gets/s=%6.3fK\n",
                                entryCount, ITERATIONS / seconds / 1e3);
                    }
                };
            for (Thread thread : threads)
                thread.start();
            for (Thread thread : threads) {
                try {
                    thread.join();
                } catch (InterruptedException e) {
                }
            }
        }
    }

}

/**
 * How to determine a Map's memory efficiency. AirConcurrentMap: reaches 39
 * bytes/Entry above about 1K Entries, CSLM 59 bytes/entry, CHashMap 65
 * bytes/Entry.
 */
class TwitterDemoForMapMemorySize {
    public static void main(String... args) {
        System.out.println("TwitterDemoForMapMemorySize");
        final int MAX_SIZE = 32768;
        Random random = new Random(1);
        // We must hold on to all the maps or they will get optimized away.
        ArrayList<Map<Long, Long>> maps = new ArrayList<Map<Long, Long>>(MAX_SIZE);
        for (long mapSize = 1, nMaps = MAX_SIZE; mapSize <= MAX_SIZE; mapSize <<= 1, nMaps >>= 1) {
            // clear sets all elements null but does not shorten internal array.
            maps.clear();
            long size0 = memorySize();
            for (int k = 0; k < nMaps; k++) {
                Map<Long, Long> map = new AirConcurrentMap<>();
                // Map<Long, Long> map = new ConcurrentHashMap<Long, Long>();
                // Map<Long, Long> map = new ConcurrentSkipListMap<Long,
                // Long>();
                maps.add(map);
                for (int m = 0; m < mapSize; m++) {
                    Long n = random.nextLong();
                    map.put(n, n);
                }
            }
            long size1 = memorySize();
            // outputting something dependent on maps.size() prevents optimizing
            // out and gc.
            long bytesPerMap = (size1 - size0) / maps.size();
            System.out.println("Entries=" + mapSize +
                    " bytes/Entry=" + bytesPerMap / mapSize + " memory=" + (size1 - size0));
        }
    }

    static long memorySize() {
        Runtime.getRuntime().gc();
        long totalMemory = Runtime.getRuntime().totalMemory();
        long freeMemory = Runtime.getRuntime().freeMemory();
        return totalMemory - freeMemory;
    }
}

/**
 * Test Iterator performance with various Maps. AirConcurrentMap wins by a
 * factor of 2 to 3.
 * 
 * <pre>
 * AirConcurrentMap:    Entries=11000000 gets/s=52172.725K
 * HashMap:             Entries=11000000 gets/s=26719.118K
 * CSLM:                Entries=11000000 gets/s=24056.478K
 * CHM:                 Entries=11000000 gets/s=22709.103K
 * TreeMap:             Entries=11000000 gets/s=16525.371K
 * </pre>
 */
class TwitterDemoForMapIterator {
    public static void main(String... args) {
        System.out.println("TwitterDemoForMapIterator");
        final int MAX_SIZE = args.length >= 1 ? Integer.valueOf(args[0]) : 10_000_000;
        final int STEP_SIZE = args.length >= 2 ? Integer.valueOf(args[1]) : 1000_000;
        Random random = new Random();
        Map<Integer, Integer> map = new AirConcurrentMap<Integer, Integer>();
        // Map<Integer, Integer> map = new HashMap<Integer, Integer>();
        // Map<Integer, Integer> map = new TreeMap<Integer, Integer>();
        // Map<Integer, Integer> map = new ConcurrentSkipListMap<Integer,
        // Integer>();
        // Map<Integer, Integer> map = new ConcurrentHashMap<Integer,
        // Integer>();
        // Without a tangible result, the loop can be optimized out!
        for (int i = 0; i < MAX_SIZE;) {
            for (int j = 0; j < STEP_SIZE; j++) {
                Integer n = new Integer(random.nextInt());
                map.put(n, n);
                i++;
            }
            long t0 = System.nanoTime();
            long total = 0;
            for (Integer k : map.keySet()) {
                total += k.intValue();
            }
            long t1 = System.nanoTime();
            double seconds = (t1 - t0) / 1e9;
            System.out.printf("Entries=%d gets/s=%6.3fK total=%d\n",
                    i, MAX_SIZE / seconds / 1e3, total);
        }
    }
}

/**
 * Test forEach() with lambdas with various Maps. AirConcurrenMap wins by a
 * multiple of 3 to 6 times.
 * 
 * The AirConcurrentMap Threaded visiting technique using a ThreadedVisitor
 * BiConsumer is not used here, but is shown in other tests to be 10 to 20 times
 * faster.
 * 
 * <pre>
 * AirConcurrentMap w/BiConsumer Entries=10000000 ops/s=115833.202K map.size()=9988406
 * 
 * AirConcurrentMap:    Entries=10000000 ops/s=40651.439K map.size()=9988406
 * HashMap:             Entries=10000000 ops/s=37160.047K map.size()=9988406
 * CSLM:                Entries=10000000 ops/s=28687.409K map.size()=9988406
 * CHM:                 Entries=10000000 ops/s=25548.065K map.size()=9988406
 * TreeMap:             Entries=10000000 ops/s=20235.311K map.size()=9988406
 * </pre>
 */
class TwitterDemoForBiConsumerWithLambdaInteger {
    static final int STEP_SIZE = 1000_000;
    // Without a tangible result, the loop can be optimized out!
    static int sum = 0;

    public static void main(String... args) {
        System.out.println("TwitterDemoForBiConsumerWithLambdaInteger");
        final int MAX_SIZE = args.length >= 1 ? Integer.valueOf(args[0]) : 10_000_000;
        Random random = new Random(315664);
        AirConcurrentMap<Integer, Integer> map = new AirConcurrentMap<Integer, Integer>();
        // Map<Integer, Integer> map = new HashMap<Integer, Integer>();
//         Map<Integer, Integer> map = new TreeMap<Integer, Integer>();
//         Map<Integer, Integer> map = new ConcurrentSkipListMap<Integer, Integer>();
//         Map<Integer, Integer> map = new ConcurrentHashMap<Integer, Integer>();
        for (int i = 0; i < MAX_SIZE;) {
            for (int j = 0; j < STEP_SIZE; j++) {
                Integer n = new Integer(random.nextInt());
                map.put(n, n);
                i++;
            }
            sum = 0;
            long t0 = System.nanoTime();
            if (false)
                map.forEach((k, v) -> sum += v.intValue());
            else
                // Fast, for AirConcurrentMap. 
                // It becomes Java-8 dependent by using a BiConsumer and lambda
                map.visit(new MapVisitorBiConsumerWrapper<Integer, Integer>(
                        (k, v) -> sum += v.intValue()));
            long t1 = System.nanoTime();
            double seconds = (t1 - t0) / 1e9;
            System.out.printf("Entries=%d ops/s=%6.3fK map.size()=%d\n", i, i / seconds / 1e3, map.size());
        }
        System.out.println("sum=" + sum);
    }
}

/**
 * Test forEach() with lambdas with various Maps. AirConcurrenMap wins by a
 * multiple of 4 to 7 times.
 * 
 * The separate scans are all done in parallel to get all cores running,
 * simulating heavy overall load. The AirConcurrentMap Threaded visiting
 * technique using a ThreadedVisitor BiConsumer is not used here, but is shown
 * in other tests to be 10 to 20 times faster.
 * 
 * <pre>
 * AirConcurrentMap:    Entries=10000000 ops/s=682427.850K sum=530270295304560
 * CSLM:                Entries=10000000 ops/s=176247.562K sum=530270295304560
 * HashMap:             Entries=10000000 ops/s=115310.715K sum=530270295304560
 * TreeMap:             Entries=10000000 ops/s=117812.730K sum=530270295304560
 * CHM:                 Entries=10000000 ops/s=102514.533K sum=530270295304560
 * </pre>
 */
class TwitterDemoForThreadedLambda {
    static final int THREADS = 8;
    static final int REPEATS = 30;

    static final AirConcurrentMap<Integer, Integer> map = new AirConcurrentMap<Integer,
            Integer>();
    // static final Map<Integer, Integer> map = new
    // ConcurrentSkipListMap<Integer, Integer>();
    // static final Map<Integer, Integer> map = new HashMap<Integer, Integer>();
    // static final Map<Integer, Integer> map = new ConcurrentHashMap<Integer,
    // Integer>();
    // static final Map<Integer, Integer> map = new TreeMap<Integer, Integer>();

    static int entryCount = 0;

    public static void main(String... args) {
        System.out.println("TwitterDemoForThreadedLambda");
        final int MAX_SIZE = args.length >= 1 ? Integer.valueOf(args[0]) : 10_000_000;
        final int STEP_SIZE = args.length >= 2 ? Integer.valueOf(args[1]) : 1000_000;
        final Random random = new Random(1);
        final AtomicLong sum = new AtomicLong();
        while (entryCount <= MAX_SIZE) {
            for (int i = 0; i < STEP_SIZE; i++) {
                Integer n = new Integer(random.nextInt());
                map.put(n, n);
                entryCount++;
            }
            Thread[] threads = new Thread[THREADS];
            for (int j = 0; j < THREADS; j++)
                threads[j] = new Thread() {
                    long localSum = 0;

                    public void run() {
                        for (int r = 0; r < REPEATS; r++) {
                            if (false) {
                                map.forEach((k, v) -> localSum += v.longValue());
                            } else if (false) {
                                // Not faster for AirConcurrentmap
                                map.forEach(new BiConsumer<Integer, Integer>() {
                                    public final void accept(Integer k, Integer v) {
                                        localSum += v.intValue();
                                    }
                                });
                            } else if (true) {
                                // The fast AirConcurrentMap.visit() is only
                                // Java 6
                                // dependent, so we adapt the
                                // visitor with a 1.8-dependent wrapper.
                                map.visit(
                                        new MapVisitorBiConsumerWrapper<Integer, Integer>(
                                                (k, v) -> localSum += v.longValue()));
                            }
                        }
                        sum.addAndGet(localSum);
                    }
                };
            sum.set(0);
            for (Thread thread : threads)
                thread.start();
            long t0 = System.nanoTime();
            for (Thread thread : threads) {
                try {
                    thread.join();
                } catch (InterruptedException e) {
                }
            }
            long t1 = System.nanoTime();
            double seconds = (t1 - t0) / 1e9;
            System.out.printf("Entries=%d ops/s=%6.3fK sum=%d\n",
                    entryCount,
                    entryCount / seconds / 1e3 * THREADS * REPEATS,
                    sum.get());
        }
    }
}

/**
 * Test forEach() with lambdas with various Maps. AirConcurrenMap uses a
 * threaded technique and it wins by a multiple of 10 to 20 times.
 * 
 * <pre>
 * AirConcurrentMap:    Entries=10000000 ops/s=358614.367K
 * HashMap:             Entries=10000000 ops/s=37201.442K
 * CSLM:                Entries=10000000 ops/s=25353.822K
 * CHM:                 Entries=10000000 ops/s=22421.307K
 * TreeMap:             Entries=10000000 ops/s=17833.000K
 * </pre>
 */
class TwitterDemoForThreadedSummerPerformance {
    static long sum = 0;

    public static void main(String... args) {
        System.out.println("TwitterDemoForThreadedSummerPerformance");
        final int MAX_SIZE = args.length >= 1 ? Integer.valueOf(args[0]) : 10_000_000;
        final int STEP_SIZE = args.length >= 2 ? Integer.valueOf(args[1]) : 1000_000;
        Random random = new Random(315664);
        Map<Long, Long> map = new AirConcurrentMap<Long,
                Long>();
        // Map<Long, Long> map = new HashMap<Long, Long>();
        // Map<Long, Long> map = new TreeMap<Long, Long>();
        // Map<Long, Long> map = new ConcurrentSkipListMap<Long, Long>();
        // Map<Long, Long> map = new ConcurrentHashMap<Long, Long>();
        for (long i = 0; i < MAX_SIZE;) {
            for (int j = 0; j < STEP_SIZE; j++) {
                Long n = new Long(random.nextLong());
                map.put(n, n);
                i++;
            }
            long t0 = System.nanoTime();
            // A special Way to parallelize AirConcurrentMap compatible with all
            // Java versions
            sum += new ThreadedSummer<Long>().getSum(map);
            // map.forEach((k, v) -> sum += v.longValue());
            long t1 = System.nanoTime();
            double seconds = (t1 - t0) / 1e9;
            System.out.printf("Entries=%d ops/s=%6.3fK\n", i, i / seconds / 1e3);
        }
        System.out.println("sum=" + sum);
    }

    /**
     * This is a re-usable general-purpose very fast map summer for AirConcurrentMap
     * 
     * @param <K>
     */
    static class ThreadedSummer<K> extends ThreadedMapVisitor<K, Long> {
        long sum;

        long getSum(Map<K, Long> map) {
            sum = 0;
            if (map instanceof VisitableMap) {
                ((VisitableMap)map).visit(this);
            } else {
                // Make this handle non-Visitable Maps as well for convenience.
                // You can use lambdas here but it becomes Java 1.8 dependent.
                for (Long v : map.values()) {
                    sum += v.longValue();
                }
                // This can be faster than iterator for some standard Maps, but
                // not CSLM
                // map.forEach((k, v) -> sum += v.longValue());
            }
            return sum;
        }

        public void visit(K key, Long value) {
            sum += value.longValue();
        }

        public ThreadedSummer<K> split() {
            return new ThreadedSummer<K>();
        }

        public void merge(ThreadedMapVisitor<K, Long> otherSummer) {
            sum += ((ThreadedSummer<K>)otherSummer).sum;
        }
    }
}

/**
 * This simple code performance tests Java 8 Stream reducers, collectors and
 * forEach(). AirConcurrentMap wins by 4.2 to 7.4 times.
 * 
 * We compare with AirConcurrentMap visit(), which uses internal threading.
 * (Intel i7 3GHz quad-core hyperthreaded)
 * 
 * <pre>
 * AirConcurrentMap forEach:    Entries=10000000 ops/s=341933.286K
 * 
 * Using stream reducers:
 * CSLM:                Entries=10000000 ops/s=81102.013K
 * AirConcurrentMap:    Entries=10000000 ops/s=77899.483K
 * TreeMap:             Entries=10000000 ops/s=58306.705K
 * HashMap:             Entries=10000000 ops/s=53881.339K
 * CHM:                 Entries=10000000 ops/s=52893.072K
 * 
 * Using the parallel stream forEach():
 * HashMap:             Entries=10000000 ops/s=50186.038K
 * CSLM:                Entries=10000000 ops/s=49186.583K
 * TreeMap:             Entries=10000000 ops/s=47517.139K
 * CHM:                 Entries=10000000 ops/s=47213.580K
 * AirConcurrentMap:    Entries=10000000 ops/s=46708.239K
 * </pre>
 */
class TwitterDemoForStreamsVsAirConcurrentMapVisit {
    static final int REPEATS = 100;

    static long sum = 0;
    static AtomicLong atomicSum = new AtomicLong();

    public static void main(String... args) {
        System.out.println("TwitterDemoStreamsVsAirConcurrentMapVisit");
        final int MAX_SIZE = args.length >= 1 ? Integer.valueOf(args[0]) : 10_000_000;
        final int STEP_SIZE = args.length >= 2 ? Integer.valueOf(args[1]) : 1000_000;
        Random random = new Random(315664);
        Map<Long, Long> map = new AirConcurrentMap<Long, Long>();
        // Map<Long, Long> map = new HashMap<Long, Long>();
        // Map<Long, Long> map = new TreeMap<Long, Long>();
        // Map<Long, Long> map = new ConcurrentSkipListMap<Long, Long>();
        // Map<Long, Long> map = new ConcurrentHashMap<Long, Long>();
        for (long i = 0; i < MAX_SIZE;) {
            for (int j = 0; j < STEP_SIZE; j++, i++) {
                Long n = new Long(random.nextLong());
                map.put(n, n);
            }
            long t0 = System.nanoTime();
            for (int repeats = 0; repeats < REPEATS; repeats++) {
                // A special reusable means to parallelize AirConcurrentMap.
                if (true)
                    sum += new ThreadedSummer<Long>().getSum(map);
                else if (false)
                    map.values().stream().parallel().forEach(x -> {
                        atomicSum.set(atomicSum.get() + x.longValue());
                    });
                else if (false)
                    sum += map.values().stream().parallel()
                            .reduce(0L, (x, y) -> x + y);
                else if (false)
                    sum += map.values().stream().parallel()
                            .collect(Collectors.summingLong(
                                    x -> ((Long)x).longValue()));
                else if (false)
                    // Slowest
                    map.forEach((k, v) -> sum += v.longValue());
            }
            long t1 = System.nanoTime();
            double seconds = (t1 - t0) / 1e9;
            System.out.printf("Entries=%d ops/s=%6.3fK\n",
                    i, i * REPEATS / seconds / 1e3);
        }
        System.out.println("sum=" + sum);
    }

    // This implementation allows AirConcurrentMap to parallelize
    static final class ThreadedSummer<K> extends ThreadedMapVisitor<K, Long> {
        // Does not need to be atomic - not shared.
        long sum;

        final long getSum(Map<K, Long> map) {
            // initalize here so a single ThreadedSummer instance can be reused.
            sum = 0;
            if (map instanceof VisitableMap) {
                ((VisitableMap)map).visit(this);
            } else {
                // Optionally make this run on non-Visitable Maps
                // Use a loop instead to make this code version-independent
                map.forEach((k, v) -> sum += v.longValue());
            }
            return sum;
        }

        @Override
        public final void visit(K key, Long value) {
            sum += value.longValue();
        }

        // For parallelism
        @Override
        public final ThreadedSummer<K> split() {
            return new ThreadedSummer<K>();
        }

        // For parallelism
        @Override
        public final void merge(ThreadedMapVisitor<K, Long> otherSummer) {
            sum += ((ThreadedSummer<K>)otherSummer).sum;
        }
    }
}

/**
 * This simple code shows how to use and performance test Java 8 Map Stream
 * Collectors to do summing. AirConcurrentMap wins at 3x to 4x.
 * 
 * We also test a wrapper class ValueCollectingThreadedMapVisitor for
 * AirConcurrentMap that uses the internal thread pool which is about 3 times to
 * 4 times faster. We use a 'collection' that is just a 'LongHolder'. (Intel i7
 * 3GHz quad-core hyperthreaded)
 * 
 * <pre>
 * AirConcurrentMap collect:    Entries=10000000 ops/s=316164.749K
 * 
 * Using regular parallel collectors:
 * 
 * CSLM:                Entries=9000000 ops/s=100714.430K
 * TreeMap:             Entries=10000000 ops/s=88388.320K
 * AirConcurrentMap:    Entries=10000000 ops/s=83344.249K
 * HashMap:             Entries=10000000 ops/s=80473.108K
 * CHM:                 Entries=10000000 ops/s=74286.026K
 * </pre>
 * 
 * sum=-8605951044782024408
 */
class TwitterDemoForStreamsCollectorsVsAirConcurrentMap {
    static final int REPEATS = 100;

    static long sum = 0;

    public static void main(String... args) {
        System.out.println("TwitterDemoForStreamsCollectorsVsAirConcurrentMap");
        final int MAX_SIZE = args.length >= 1 ? Integer.valueOf(args[0]) : 10_000_000;
        final int STEP_SIZE = args.length >= 2 ? Integer.valueOf(args[1]) : 1000_000;
        Random random = new Random(315664);
        Map<Long, Long> map = new AirConcurrentMap<Long, Long>();
        // Map<Long, Long> map = new HashMap<Long, Long>();
        // Map<Long, Long> map = new TreeMap<Long, Long>();
        // Map<Long, Long> map = new ConcurrentSkipListMap<Long, Long>();
        // Map<Long, Long> map = new ConcurrentHashMap<Long, Long>();
        for (long i = 0; i < MAX_SIZE;) {
            for (int j = 0; j < STEP_SIZE; j++) {
                Long n = new Long(random.nextLong());
                map.put(n, n);
                i++;
            }
            long t0 = System.nanoTime();
            for (int repeats = 0; repeats < REPEATS; repeats++) {
                // @formatter:off
                if (true) {
                    // Very fast internally threaded adapter that bundles the 1.8 dependency.
                    // AirConcurrentMap: 3x is typical. AirConcurrentMap itself is 1.6 dependent.
                    sum += new ValueCollectingThreadedMapVisitor<Long, Long, LongHolder>( 
                        LongHolder::new, (r, v) -> { r.x += v; },
                        (r1, r2) -> { r1.x += r2.x; }).getResult(map).x;
                }
                if (false) {
                    // This way adds a 'mapper' lambda as the first parameter.
                    sum += new CollectingThreadedMapVisitor<Long, Long, LongHolder, Long>(
                        (k, v) -> v,
                        LongHolder::new,
                        (r, v) -> { r.x += v; }, 
                        (r1, r2) -> { r1.x += r2.x; }).getResult(map).x;
                }
                if (false) {
                    // standard technique
                    sum += map.values().stream().parallel().collect(
                        LongHolder::new, (r, v) -> { r.x += v; }, 
                        (r1, r2) -> { r1.x += r2.x; }).x;
                }
                // @formatter:on
                // The standard library has a specialized summer:
                if (false)
                    sum += map.values().stream().parallel().collect(
                            Collectors.summingLong(x -> ((Long)x).longValue()));
            }
            long t1 = System.nanoTime();
            double seconds = (t1 - t0) / 1e9;
            System.out.printf("Entries=%d ops/s=%6.3fK\n",
                    i, i * REPEATS / seconds / 1e3);
        }
        System.out.println("sum=" + sum);
    }

    // The 'container' we work with for dealing with longs, such as for summing.
    static class LongHolder {
        public long x;
    }
}

/**
 * This simple code shows how to use and test Java 8 Map Stream Collectors with
 * ArrayLists. AirConcurrenMap wins with its internal threads.
 * 
 * There is a collector class that uses AirConcurrentMap's internal thread pool
 * for extra speed. (Intel i7 3GHz quad-core hyperthreaded)
 * 
 * <pre>
 * AirConcurrentMap collectParallel(): Entries=10000000 ops/s=68928.258K size=10000000   
 * 
 * Using regular non-native parallel collectors (ACM in serial mode)
 * 
 * TreeMap:             Entries=10000000 ops/s=43153.720K size=10000000
 * HashMap:             Entries=10000000 ops/s=42291.941K size=10000000
 * CHM:                 Entries=10000000 ops/s=41806.186K size=10000000
 * AirConcurrentMap:    Entries=10000000 ops/s=39429.624K size=10000000
 * CSLM:                Entries=10000000 ops/s=31816.849K size=10000000
 * </pre>
 * 
 * sum=-5416378075002339540 (Intel i7 3GHz quad-core hyperthreaded)
 */
class TwitterDemoForStreamsCollectorsVsAirConcurrentMapArrayList {
    // static final int REPEATS = 300;
    static final int REPEATS = 30;

    public static void main(String... args) {
        System.out.println("TwitterDemoForStreamsCollectorsVsAirConcurrentMapArrayList");
        final int MAX_SIZE = args.length >= 1 ? Integer.valueOf(args[0]) : 10_000_000;
        final int STEP_SIZE = args.length >= 2 ? Integer.valueOf(args[1]) : 1000_000;
        Random random = new Random(315664);
        // Uncomment a Map to test.
        Map<Long, Long> map = new AirConcurrentMap<Long, Long>();
        // Map<Long, Long> map = new HashMap<Long, Long>();
        // Map<Long, Long> map = new TreeMap<Long, Long>();
        // Map<Long, Long> map = new ConcurrentSkipListMap<Long, Long>();
        // Map<Long, Long> map = new ConcurrentHashMap<Long, Long>();
        for (long i = 0; i < MAX_SIZE;) {
            for (int j = 0; j < STEP_SIZE; j++) {
                Long n = new Long(random.nextLong());
                map.put(n, n);
                i++;
            }
            long t0 = System.nanoTime();
            ArrayList<Long> result = null;
            for (int k = 0; k < REPEATS; k++) {
                if (true) {
                    // Very fast internally threaded adapter that bundles the
                    // 1.8 dependency.
                    // AirConcurrentMap: 3x is typical. AirConcurrentMap itself
                    // is 1.6 dependent.
                    result = new ValueCollectingThreadedMapVisitor<Long, Long, ArrayList>(
                            ArrayList::new, ArrayList::add, ArrayList::addAll).getResult(map);
                } else if (false) {
                    result = map.values().stream().collect(
                            ArrayList::new, ArrayList::add, ArrayList::addAll);
                } else if (false) {
                    // AirConcurrentMap has a bug here - it is slow and runs
                    // out of memory in parallel mode, but is still fast in
                    // serial mode, even compared with parallel for the others.
                    result = map.values().stream().parallel().collect(
                            ArrayList::new, ArrayList::add, ArrayList::addAll);
                }
            }
            long t1 = System.nanoTime();
            double seconds = (t1 - t0) / 1e9;
            System.out.printf("Entries=%d ops/s=%6.3fK size=%d\n",
                    i, i * REPEATS / seconds / 1e3, result.size());
        }
    }

}