Executor Framework

• If you have thousands of task to be executed and if you create each thread for thousands of tasks, you will get performance overheads as creation and maintenance of each thread is an overhead.

• Executor framework solves this problem.

• Using executor framework, you can create specified number of threads and reuse them to execute more tasks once it completes its current task.

• It simplifies the design of creating multithreaded application and manages thread life cycles.

• The programmer does not have to create or manage threads themselves, that’s the biggest advantage of executor framework.

• It provides a high level concurrency APIs.

Interfaces of Executor framework

1. java.util.concurrent.Executor

   • This interface provides a way of decoupling task submission from the mechanics of how each task will be run, including details of thread use, scheduling, etc. An Executor is normally used instead of explicitly creating threads.

   • Executor has only single method. Definition is given below,

     •   public interface Executor {
             void execute(Runnable command);
         }

2. java.util.concurrent.ExecutorService

   • An Executor that provides methods to manage termination and methods that can produce a Future for tracking progress of one or more asynchronous tasks.

   • An ExecutorService can be shut down, which will cause it to reject new tasks. Two different methods are provided for shutting down an ExecutorService

   • The shutdown() method will allow previously submitted tasks to execute before terminating, while the shutdownNow() method prevents waiting tasks from starting and attempts to stop currently executing tasks. Upon termination, an executor has no tasks actively executing, no tasks awaiting execution, and no new tasks can be submitted. An unused ExecutorService should be shut down to allow reclamation of its resources.

   • APIs of ExecutorService

     • submit() methods allow to submit a runnable or callable task.

           <T> Future<T> submit(Callable<T> task);
           Future<?> submit(Runnable task);
           <T> Future<T> submit(Runnable<T> task, T result);

       • Throws java.util.concurrent.RejectedExecutionException, if task cannot be scheduled for execution.

     • shutdown() methods allow to requests shutdown of submitted tasks. This method only initiates a shutdown.

          void shutdown()
          List<Runnable> shutdownNow()

       • No new tasks will be accepted, on calling again on an executor service which has already shutdown, this method does nothing.

       • shutdownNow method, attempts to stop all actively executing tasks, halts the processing of waiting tasks, and returns a list of the tasks that were awaiting execution but were not executing. However, shutdown() intiates shutdown, making sure all task in execution and the ones in task queue are completed.

     • awaitTermination() blocks untill all tasks have completed execution after a shutdown request, or the timeout occurs, or the current thread is interrupted, whichever happens first.

           boolean awaitTermination(long timeout, TimeUnit unit)throws InterruptedException

       • Returns true, if all submitted tasks are terminated and false if time elapsed before termination.

     • invokeAll(), executes the given tasks, returning a list of Futures holding their status and results once completed.

           <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks);
       
           <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit);
       
           <T> T invokeAny(Collection<? extends Callable<T>> tasks);
       
           <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)

       • The timeout version of these APIs, waits for maximum time to wait. After the wait time is over and a task is not completed then that task/s will be cancelled.

       • Note that the collection passed to the methods should not be altered after this method is in progress. Doing so, may result in undefined behavior.

       • These family of methods throw java.lang.InterruptedException in case the thread is interruped while waiting, and unfinished tasks are cancelled.

       • invokeAny(), executes the given tasks, returning the result of one that has completed successfully, if any task completed.

       • invokeAny() which takes timeout as argument may throw TimeoutException if none of the task completed in the given time.

3. java.util.concurrent.ScheduledExecutorService

   • A Sub interface of ExecutorService that can schedule commands to run after a given delay, or to execute periodically.

   • The scheduleAtFixedRate and scheduleWithFixedDelay methods create and execute tasks that run periodically until cancelled.

         <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit);
         
         ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit);
         
         ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit);
         
         ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit);

ThreadPoolExecutor and its Type

ThreadPoolExecutor

• Thread pools executor helps to execute asynchronous task without managing thread related api.

• It abstracts away thread management from developer.

• There are different types of thread pools which can be obtained by using factory methods of Executors class.

• These factory methods are specialised fine tuned thread pools.

• Important terms and parameters passed while creating a thread pool.

  • core pool size : The number of threads to keep in the pool, even if they are idle, (unless allowCoreThreadTimeOut is set, false by default).

  • maximum pool size : The maximum number of threads to allow in the pool.

  • Keep alive time : When the number of threads is greater than the core, this is the maximum time that excess idle threads will wait for new tasks before terminating.

  • Work Queue : The queue to use for holding tasks before they are executed. This queue will hold only the tasks submitted by the execute method.

    • Strategies for queuing,

      • Direct hand off, uses Synchronous queue.

      • Unbounded queue, uses LinkedBlocking queue.

      • Bounded queue, uses ArrayBlocking queue.

  • RejectionExecution Handler : The handler to use when execution is blocked because the thread bounds and queue capacities are reached.

  • Thread Factory : The factory to use when executor needs to create a new thread.

  • Rejection Policy : New tasks submitted in method execute (Runnable) will be rejected when the,

    • Executor has been shut down

    • Executor uses finite bounds for both maximum threads and work queue capacity, and is saturated.

    • Rejection policies supported are,

      • Abort policy, submitting new task throws RejectedExecutionException.

      • Caller Runs policy, submitting new task will execute the task on the caller thread itself. This slows down the task submission rate.

      • Discard policy, submission of new task silently discards it.

      • Discard oldest policy, submission of new task drops the existing oldest task and itself is added to queue.

ScheduledThreadPoolExecutor

• A ThreadPoolExecutor that can additionally schedule commands to run after a given delay, or to execute periodically. This class is preferable to Timer when multiple worker threads are needed, or when the additional flexibility or capabilities of ThreadPoolExecutor (which this class extends) are required.

Types of thread pool

Type	Core size	Maximum size	Queue used	Keep Alive	Remarks
FixedThreadPool	n	same as core pool size	java.util.concurrent.LinkedBlockingQueue	NA (0 seconds)	Unlimited task can be submitted, but limited pool size
CachedThreadPool	0	Integer.MAX_VALUE	java.util.concurrent.SynchronousQueue	60 seconds	Queue can accomodate only one task, but supports unbounded threads in thread pool.
ScheduledThreadPool	n	Integer.MAX_VALUE	java.util.concurrent.DelayQueue	60 seconds	Special queue, which returns task based on scheduling time of task.
SingleThreadExecutor	1	1	java.util.concurrent.LinkedBlockingQueue	0 seconds	Unbounded queue, and single thread

Thread factory

• An Interface which defines behavior for new thread creation.

• An object that creates new threads on demand. Using thread factories removes hardwiring of calls to new Thread, enabling applications to use special thread subclasses, priorities, etc.

      public interface ThreadFactory {
          Thread newThread(Runnable r);
      }

• Executors class has a defaultThreadFactory() that creates new threads. These thread are non-daemon, with Priority set to NORM_PRIORITY.

Fork-Join Framework

• The fork/join framework is an implementation of the ExecutorService interface that helps you take advantage of multiple processors.

• It is designed for work that can be broken into smaller pieces recursively. The goal is to use all the available processing power to enhance the performance of your application.

• It provides a high performance, fine-grained task execution for java data parallelism.

• This framework is used by Arrays.parallelSort(), CompletableFuture, parallelstream().

• This framework supports divide and conquer parallelism, as mentioned below,

    if (my portion of the work is small enough):

        do the work directly
    else:
        split my work into two pieces
        invoke the two pieces and wait for the results

ForkJoinPool

• Working of ForkJoinPool.

  • Unlike ThreadPoolExecutor, which has single shared blocking queue from which task can be assigned to worker thread in the pool, ForkJoinPool has shared as well as thread-specific queue (actually deque) also known as Work-stealing queue. (Internally these work-stealing queues are implemented by a Java class named WorkQueue).

  • New task are submitted to a shared queue of fork-join pool. This shared queue is accessed by fork-join pool and pull the task to these work-stealing queue.

  • The goal of work-stealing queue is to maximize processor core utilization.

  • These threads only halts/blocks when there is absolutely no tasks to be executed.

  • To make sure that absolutely no work is there to be executed, these worker threads check for queues operated by other threads (i.e., other worker thread's work-stealing queues) so as to find work to execute. Hence, stealing the work from other queue and helping out threads which are busy. Note that task are taken from tail of the queue and not the head of the queue (i.e., stealing is done in FIFO order).

  • The stealing is done from a random thread queue, in order to avoid contention.

  • If a task run by worker thread, calls fork() the new task created is pushed to that worker's queue. These queues uses LIFO (last in first out) policy to process the tasks in the queue. The LIFO policy is used to take advantage of locality of reference and cache performance.

• Characteristics of ForkJoinPool

  • By default, size is equal to number of available processors. However, size can be controlled by specifying the parallelism while constructing object.

  • The common fork join pool can be accessed via api.

        static ForkJoinPool commonPool()

    • This pool is the default pool used if not explicitly supplied while construction.

• APIs of ForkJoinPool

  • Eventhough this implements ExecutorService apis, which allows submission of Runnable and Callable tasks, these methods does not leverage the powerful feature of ForkJoinPool.

        void execute(Runnable task)
        <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
    
        <T> ForkJoinTask<T> submit(Callable<T> task)
        ForkJoinTask<?> submit(Runnable task)
        <T> ForkJoinTask<T> submit(Runnable task, T result)

  • It implements methods which take ForkJoinTask, which enables us to make use of the worker threads of ForkJoinPool. These methods enable us to use the powerful feature of ForkJoinPool.

        void execute(ForkJoinTask) 	// async execution, no result, only side effects. No blocking
        <T> T invoke(ForkJoinTask)  // Awaits and returns result, hence blocking
        <T> ForkJoinTask<T> submit(ForkJoinTask) // Arrange execution and obtain Future

ForkJoinTask

• A ForkJoinTask is a thread-like entity that is much lighter weight than a normal thread.

• It is a light-weight form of Future.

• Huge numbers of tasks and subtasks may be hosted by a small number of actual threads in a ForkJoinPool, at the price of some usage limitations.

• The primary coordination mechanisms are fork(), that arranges asynchronous execution, and join(), that doesn't proceed until the task's result has been computed.

• Computations should ideally avoid synchronized methods or blocks, and should minimize other blocking synchronization apart from joining other tasks or using synchronizers such as Phasers that are advertised to cooperate with fork/join scheduling.

• The ForkJoinTask class is not usually directly subclassed. Instead, you subclass one of the abstract classes that support a particular style of fork/join processing, typically RecursiveAction for most computations that do not return results, RecursiveTask for those that do, and CountedCompleter for those in which completed actions trigger other actions.

• Normally, a concrete ForkJoinTask subclass declares fields comprising its parameters, established in a constructor, and then defines a compute method that somehow uses the control methods supplied by this base class.