src/test/java/net/onrc/onos/core/datastore/hazelcast/HZDistributedAtomicLongTest.java

package net.onrc.onos.core.datastore.hazelcast;

import static org.junit.Assert.*;

import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import net.onrc.onos.core.datastore.ObjectDoesntExistException;
import net.onrc.onos.core.util.distributed.DistributedAtomicLong;

import org.junit.Before;
import org.junit.Test;

/**
 * Test cases for HZDistributedAtomicLong.
 */
public class HZDistributedAtomicLongTest {
    static {
        // configuration to quickly fall back to instance mode for faster test run
        System.setProperty("net.onrc.onos.core.datastore.hazelcast.client.attemptLimit", "0");
    }

    static final String TEST_COUNTER_NAME = "Counter" + UUID.randomUUID();

    DistributedAtomicLong counter;

    @Before
    public void setUp() throws Exception {
        counter = new HZDistributedAtomicLong(
                    HZClient.getClient(), TEST_COUNTER_NAME);
        counter.set(0L);
    }

    @Test
    public void testGet() {
        assertEquals(0L, counter.get());
    }

    @Test
    public void testAddAndGet() {
        assertEquals(0L, counter.get());
        assertEquals(1L, counter.addAndGet(1L));
        assertEquals(3L, counter.addAndGet(2L));
        assertEquals(7L, counter.addAndGet(4L));
    }

    @Test
    public void testSet() {
        counter.set(42L);
        assertEquals(42L, counter.get());
    }

    @Test
    public void testIncrementAndGet() {
        assertEquals(1L, counter.incrementAndGet());
        assertEquals(2L, counter.incrementAndGet());
        assertEquals(3L, counter.incrementAndGet());
    }

    /**
     * Callable task incrementing atomicLong.
     */
    private static final class AdderTask implements Callable<Long> {
        // using Map as Set
        private final ConcurrentMap<Long, Long> uniquenessTestSet;
        // Queue given here should be Thread-safe
        private final Queue<Long> incrementTimes;
        private final HZDistributedAtomicLong counter;
        private final int numIncrements;

        /**
         * Constructor.
         *
         * @param numIncrements number of increments to execute
         * @param uniquenessTestSet ThreadSafe Map to store increment result
         * @param incrementTimes ThreadSafe Queue to store time it
         *      took on each increment
         */
        public AdderTask(int numIncrements,
                ConcurrentMap<Long, Long> uniquenessTestSet,
                Queue<Long> incrementTimes) {

            super();
            this.uniquenessTestSet = uniquenessTestSet;
            this.incrementTimes = incrementTimes;
            this.counter = new HZDistributedAtomicLong(
                    HZClient.getClient(), TEST_COUNTER_NAME);
            this.numIncrements = numIncrements;
        }

        @Override
        public Long call() throws ObjectDoesntExistException {
            for (int i = 0; i < numIncrements; ++i) {
                final long start = System.nanoTime();
                final long incremented = counter.addAndGet(1L);
                incrementTimes.add(System.nanoTime() - start);
                final Long expectNull = uniquenessTestSet.putIfAbsent(
                                                    incremented, incremented);
                assertNull(expectNull);
            }
            return null;
        }
    }

    private static final int NUM_THREADS = Integer.parseInt(
                              System.getProperty(
                                      "HZDistributedAtomicLongTest.NUM_THREADS",
                                        System.getProperty("NUM_THREADS",
                                                           "10")));

    private static final int NUM_INCREMENTS = Integer.parseInt(
                              System.getProperty(
                                      "HZDistributedAtomicLongTest.NUM_INCREMENTS",
                                        System.getProperty("NUM_INCREMENTS",
                                                           "100")));

    /**
     * Increment using multiple threads to test addition is atomic.
     */
    @Test
    public void testConcurrentAddAndGet() throws InterruptedException, ExecutionException {
        ExecutorService executor = Executors.newFixedThreadPool(NUM_THREADS);

        // using Map as Set
        ConcurrentMap<Long, Long> uniquenessTestSet = new ConcurrentSkipListMap<>();
        Queue<Long> incrementTimes = new ConcurrentLinkedQueue<>();

        // Start NUM_THREADS threads and increment NUM_INCREMENTS times each
        List<Callable<Long>> tasks = new ArrayList<>(NUM_THREADS);
        for (int i = 0; i < NUM_THREADS; ++i) {
            tasks.add(new AdderTask(NUM_INCREMENTS,
                                      uniquenessTestSet,
                                      incrementTimes));
        }
        List<Future<Long>> futures = executor.invokeAll(tasks);

        // wait for all tasks to end
        for (Future<Long> future : futures) {
            future.get();
        }


        assertEquals(NUM_THREADS * NUM_INCREMENTS , uniquenessTestSet.size());

        // check uniqueness of result
        long prevValue = 0;
        for (Long value : uniquenessTestSet.keySet()) {
            assertEquals((prevValue + 1), value.longValue());
            prevValue = value;
        }

        long max = 0L;
        long min = Long.MAX_VALUE;
        long sum = 0L;
        for (Long time : incrementTimes) {
            sum += time;
            max = Math.max(max, time);
            min = Math.min(min, time);
        }
        System.err.printf("incrementCounter: th, incs, tot_incs,"
                        + " avg(ns), min(ns), max(ns), T-put(1s/avg)\n");
        System.err.printf("incrementCounter: %d, %d, %d,"
                        + " %f, %d, %d, %f\n",
                          NUM_THREADS, NUM_INCREMENTS, incrementTimes.size(),
                          sum / (double) incrementTimes.size(), min, max,
                          Math.pow(10, 9) * incrementTimes.size() / sum);

        executor.shutdown();
    }

}