CallerRunsPolicyTest.java

package com.omgitsmgp;

import java.time.Duration;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

/**
 * Serial execution with no thread pool is expected to take 10 * 2s = 20s.
 * 
 * With one thread:
 *   Running task 1 on thread main
 *   Running task 0 on thread pool-1-thread-1
 *   Running task 2 on thread main
 *   Running task 4 on thread main
 *   Running task 3 on thread pool-1-thread-1
 *   Running task 5 on thread main
 *   Running task 7 on thread main
 *   Running task 6 on thread pool-1-thread-1
 *   Running task 9 on thread main
 *   Running task 8 on thread pool-1-thread-1
 *   Completed 10 tasks with 1 threads in PT12.055748311S
 *
 * With two threads:
 *   Running task 0 on thread pool-1-thread-1
 *   Running task 2 on thread main
 *   Running task 1 on thread pool-1-thread-2
 *   Running task 3 on thread main
 *   Running task 6 on thread main
 *   Running task 4 on thread pool-1-thread-1
 *   Running task 5 on thread pool-1-thread-2
 *   Running task 8 on thread pool-1-thread-1
 *   Running task 7 on thread pool-1-thread-2
 *   Running task 9 on thread main
 *   Completed 10 tasks with 2 threads in PT8.035820264S
 *
 * With five threads:
 *   Running task 2 on thread pool-1-thread-3
 *   Running task 3 on thread pool-1-thread-4
 *   Running task 5 on thread main
 *   Running task 4 on thread pool-1-thread-5
 *   Running task 1 on thread pool-1-thread-2
 *   Running task 0 on thread pool-1-thread-1
 *   Running task 6 on thread main
 *   Running task 7 on thread pool-1-thread-3
 *   Running task 8 on thread pool-1-thread-4
 *   Running task 9 on thread pool-1-thread-1
 *   Completed 10 tasks with 5 threads in PT6.031993718S
 */
public class CallerRunsPolicyTest {
    public static void main(String[] args) {
        var startTime = System.nanoTime();

        int numThreads = 5;
        int numTasks = 10;
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
            numThreads,
            numThreads,
            10, TimeUnit.MINUTES, // keepalive
            new SynchronousQueue<>(), // default choice
            Executors.defaultThreadFactory(), // default choice
            // if all background threads are taken, execute it synchronously
            new ThreadPoolExecutor.CallerRunsPolicy());

        var futures = IntStream.range(0, numTasks)
            .mapToObj(Task::new)
            .map(executor::submit)
            .collect(Collectors.toList());
        futures.forEach(f -> {
            try {
                f.get();
            } catch (InterruptedException | ExecutionException e) {
                throw new RuntimeException(e);
            }
        });

        var endTime = System.nanoTime();
        Duration elapsed = Duration.ofNanos(endTime - startTime);
        System.out.println(
            String.format("Completed %s tasks with %s threads in %s", numTasks, numThreads, elapsed));

        executor.shutdown();
    }

    static class Task implements Runnable {
        private final int i;

        Task(int i) {
            this.i = i;
        }

        @Override
        public void run() {
            try {
                System.out.println(
                    String.format(
                        "Running task %s on thread %s", i, Thread.currentThread().getName()));
                Thread.sleep(2_000);
            } catch (InterruptedException e) {
                throw new RuntimeException();
            }
        }
    }
}