context switching tests.

gistfile1.txt

import lombok.AllArgsConstructor;
import org.junit.Test;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

public class ParallelismTest {

    private static final int MAX_ROUNDS = 1000;
    private static final int ROUND_SIZE = 100000;

    @Test
    public void trySameThreadTest() throws ClassNotFoundException, ExecutionException, InterruptedException {
        // runs all in main thread, sequentially
        CompletableFuture f = CompletableFuture.runAsync(() -> sleep(1, 0));
        f.thenRun(() -> System.out.println(Thread.currentThread().getName()));
        f.get();
    }

    @Test
    public void trySequentialTest() throws ClassNotFoundException, ExecutionException, InterruptedException {
        // runs all in main thread, sequentially
        Executor pool = command -> command.run();
        tryIncrease(pool, false);
    }

    @Test
    public void trySequentialSwitchingThreadTest() throws ClassNotFoundException, ExecutionException, InterruptedException {
        // Tries to ensure a fast voluntary context switch (access a resource)
        ForkJoinPool pool = new ForkJoinPool(1);
        tryIncrease(pool, true);
    }

    @Test
    public void tryConcurrentTest() throws ClassNotFoundException, ExecutionException, InterruptedException {
        // Tries to ensure a fast voluntary context switch (access a resource)
        ForkJoinPool pool = ForkJoinPool.commonPool();
        tryIncrease(pool, true);
    }

    /**
     * Interesting results:
     *   Using long acum ({@see LongContainer}) correctness is only warrantied by {@see trySequentialTest}
     *   Atomic is necessary for {@see trySequentialSwitchingThreadTest} too. As even synchronizing different threads to run sequentially
     *     can increase (++) on stale values due to bad timing on thread suspension.
     *
     *   When test is too long (~4 seconds to run)  {@see trySequentialTest} behaves similar to {@see trySequentialSwitchingThreadTest} due
     *     to involuntary context switch.
     *
     *  threadNames updated status is seldom visible from main thread, when updated from worker thread.
     *
     * @param pool
     * @param runConcurrentThreads
     * @throws ClassNotFoundException
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public void tryIncrease(Executor pool, boolean runConcurrentThreads) throws ClassNotFoundException, ExecutionException, InterruptedException {
        long startTimeNano = System.nanoTime();
        StringBuffer threadNames = new StringBuffer();
        AtomicLong acum = new AtomicLong(0L);
        runRounds(acum, pool, threadNames);
        if (runConcurrentThreads) {
            // We didn't wait for each task, wait for the pool then.
            ((ForkJoinPool) pool).awaitQuiescence(1, TimeUnit.DAYS);
        }

        System.out.println(threadNames.toString());
        System.out.println(String.format("With pool %s \n => Took: %,d nanos \n accum: %,d, \n ",
                pool,
                (System.nanoTime() - startTimeNano),
                acum.get()));
    }

    private void runRounds(AtomicLong accum, Executor pool, final StringBuffer threadNames)
            throws ExecutionException, InterruptedException {
        final AtomicBoolean is1stRound = new AtomicBoolean(true);
        for (int round = 0; round  < MAX_ROUNDS; round ++) {
            pool.execute(() -> increase(accum, threadNames, is1stRound.get()));
            is1stRound.compareAndSet(true, false);
        }
    }

    private void increase(AtomicLong accum, final StringBuffer threadNames, boolean dumpName) {
        if (dumpName) {
            threadNames.append(Thread.currentThread().getName());
            threadNames.append("\n");
        }
        for (int i = 0; i < ROUND_SIZE; i++) {
            sleep(0L, 0);

            accum.incrementAndGet();
        }
    }

    private void sleep (long millis, int nanos) {
        if (millis <= 0L || (millis <= 0L && nanos <= 0)) {
            return;
        }
        try {
            Thread.sleep(millis, nanos);
        }  catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @AllArgsConstructor(staticName = "of")
    private static class LongContainer {
        long theLong;
    }
}