AbstractExecutorService Foundation

The ThreadPoolExecutor inherits from AbstractExecutorService, which provides default implementations for ExecutorService interface methods. This abstract base handles task submission utilities including submit, invokeAny, and invokeAll by wrapping tasks into Future objects.

The submission mechanism follows the Template Method pattern:

public <T> Future<T> submit(Runnable task, T result) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<T> futureTask = createFutureWrapper(task, result);
    execute(futureTask);
    return futureTask;
}

protected <T> RunnableFuture<T> createFutureWrapper(Runnable runnable, T value) {
    return new FutureTask<T>(runnable, value);
}

The execute method remains abstract, forcing subclasses like ThreadPoolExecutor to define concrete execution behavior.

Constructor Parameters and Capacity Planning

ThreadPoolExecutor exposes a comprehensive constructor for fine-grained control:

public ThreadPoolExecutor(
    int minPoolSize,          // Core thread count
    int maxPoolSize,          // Maximum thread count  
    long idleTimeout,         // Keep-alive time for idle threads
    TimeUnit timeUnit,        // Unit for keep-alive time
    BlockingQueue<Runnable> taskQueue,      // Work queue
    ThreadFactory threadProducer,           // Thread creation factory
    RejectionHandler saturationHandler      // Policy when saturated
) {
    if (minPoolSize < 0 || maxPoolSize <= 0 || maxPoolSize < minPoolSize || idleTimeout < 0)
        throw new IllegalArgumentException();
    if (taskQueue == null || threadProducer == null || saturationHandler == null)
        throw new NullPointerException();
    
    this.minPoolSize = minPoolSize;
    this.maxPoolSize = maxPoolSize;
    this.taskQueue = taskQueue;
    this.idleTimeout = timeUnit.toNanos(idleTimeout);
    this.threadProducer = threadProducer;
    this.saturationHandler = saturationHandler;
}

The pool maintains a logical division between core threads (persistent workers) and overflwo threads (temporary workers up to maxPoolSize). When tasks arrive, the pool first attempts to assign them to core threads, then queues them, and finally creates overflow threads if the queue saturates.

State Management and Worker Tracking

Composite State Field

ThreadPoolExecutor packs thread count and operasional state into a single AtomicInteger called controlState:

private final AtomicInteger controlState = new AtomicInteger(encodeState(RUNNING, 0));
private static final int COUNT_BITS = Integer.SIZE - 3;
private static final int MAX_WORKERS = (1 << COUNT_BITS) - 1;

// State encodings (high 3 bits)
private static final int RUNNING    = -1 << COUNT_BITS;  // Accept new tasks, process queued tasks
private static final int SHUTDOWN   =  0 << COUNT_BITS;  // Reject new tasks, process queued tasks  
private static final int STOP       =  1 << COUNT_BITS;  // Reject new tasks, interrupt running tasks
private static final int TIDYING    =  2 << COUNT_BITS;  // All tasks terminated, pre-termination
private static final int TERMINATED =  3 << COUNT_BITS;  // Termination complete

private static int encodeState(int state, int workerCount) {
    return state | workerCount;
}

private static int getWorkerCount(int encoded) {
    return encoded & MAX_WORKERS;
}

private static int getRunState(int encoded) {
    return encoded & ~MAX_WORKERS;
}

Worker Thread Abstraction

Each worker is encapsulated in a Worker inner class extending AbstractQueuedSynchronizer to implement simple locking:

private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
    final Thread thread;
    Runnable initialTask;
    volatile long tasksProcessed;

    Worker(Runnable firstTask) {
        setState(-1);  // Pre-lock state prevents interruption during construction
        this.initialTask = firstTask;
        this.thread = threadProducer.newThread(this);
    }

    public void run() {
        processTasks(this);
    }

    protected boolean tryAcquire(int unused) {
        if (compareAndSetState(0, 1)) {
            setExclusiveOwnerThread(Thread.currentThread());
            return true;
        }
        return false;
    }

    protected boolean tryRelease(int unused) {
        setExclusiveOwnerThread(null);
        setState(0);
        return true;
    }

    void interruptIfActive() {
        Thread t;
        if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
            try { t.interrupt(); } catch (SecurityException ignored) {}
        }
    }
}

Workers are stored in a HashSet<Worker> called activeWorkers.

Task Execution Flow

Primary Entry Point

The execute method orchestrates task admission:

public void execute(Runnable command) {
    if (command == null) throw new NullPointerException();
    
    int state = controlState.get();
    
    // Phase 1: Try to add to core pool
    if (getWorkerCount(state) < minPoolSize) {
        if (spawnWorker(command, true)) return;
        state = controlState.get();
    }
    
    // Phase 2: Try to enqueue
    if (isRunning(state) && taskQueue.offer(command)) {
        int recheck = controlState.get();
        if (!isRunning(recheck) && removeFromQueue(command)) {
            saturationHandler.rejectedExecution(command, this);
        } else if (getWorkerCount(recheck) == 0) {
            spawnWorker(null, false);
        }
    }
    // Phase 3: Try to add to overflow pool
    else if (!spawnWorker(command, false)) {
        saturationHandler.rejectedExecution(command, this);
    }
}

Worker Creation Logic

private boolean spawnWorker(Runnable firstTask, boolean isCore) {
    retry:
    for (;;) {
        int state = controlState.get();
        int runState = getRunState(state);
        
        // Check if shutdown conditions prevent new workers
        if (runState >= SHUTDOWN && 
            !(runState == SHUTDOWN && firstTask == null && !taskQueue.isEmpty())) {
            return false;
        }
        
        for (;;) {
            int workerCount = getWorkerCount(state);
            if (workerCount >= MAX_WORKERS || 
                workerCount >= (isCore ? minPoolSize : maxPoolSize)) {
                return false;
            }
            
            if (incrementWorkerCount(state)) break retry;
            state = controlState.get();
            if (getRunState(state) != runState) continue retry;
        }
    }
    
    boolean started = false;
    boolean added = false;
    Worker w = null;
    try {
        w = new Worker(firstTask);
        final Thread t = w.thread;
        if (t != null) {
            final ReentrantLock lock = this.mainLock;
            lock.lock();
            try {
                int rs = getRunState(controlState.get());
                if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) {
                    if (t.isAlive()) throw new IllegalThreadStateException();
                    activeWorkers.add(w);
                    int poolSize = activeWorkers.size();
                    if (poolSize > largestPoolSize) largestPoolSize = poolSize;
                    added = true;
                }
            } finally {
                lock.unlock();
            }
            if (added) {
                t.start();
                started = true;
            }
        }
    } finally {
        if (!started) deregisterWorker(w);
    }
    return started;
}

Task Processing Loop

final void processTasks(Worker w) {
    Thread currentThread = Thread.currentThread();
    Runnable task = w.initialTask;
    w.initialTask = null;
    w.unlock();
    boolean abruptTermination = true;
    
    try {
        while (task != null || (task = fetchTask()) != null) {
            w.lock();
            
            // Ensure interruption policy compliance
            if ((getRunState(controlState.get()) >= STOP || 
                (Thread.interrupted() && getRunState(controlState.get()) >= STOP)) 
                && !currentThread.isInterrupted()) {
                currentThread.interrupt();
            }
            
            try {
                beforeExecute(currentThread, task);
                Throwable thrown = null;
                try {
                    task.run();
                } catch (RuntimeException x) {
                    thrown = x; throw x;
                } catch (Error x) {
                    thrown = x; throw x;
                } catch (Throwable x) {
                    thrown = x; throw new Error(x);
                } finally {
                    afterExecute(task, thrown);
                }
            } finally {
                task = null;
                w.tasksProcessed++;
                w.unlock();
            }
        }
        abruptTermination = false;
    } finally {
        terminateWorker(w, abruptTermination);
    }
}

Task Retrieval with Timeout

private Runnable fetchTask() {
    boolean timedOut = false;
    
    for (;;) {
        int state = controlState.get();
        int runState = getRunState(state);
        int activeCount = getWorkerCount(state);
        
        // Check termination conditions
        if (runState >= SHUTDOWN && (runState >= STOP || taskQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }
        
        boolean shouldTimeOut = allowCoreThreadTimeOut || activeCount > minPoolSize;
        
        // Check for excess workers after dynamic pool resizing
        if ((activeCount > maxPoolSize || (shouldTimeOut && timedOut)) 
            && (activeCount > 1 || taskQueue.isEmpty())) {
            if (decrementWorkerCount()) return null;
            continue;
        }
        
        try {
            Runnable r = shouldTimeOut ?
                taskQueue.poll(idleTimeout, TimeUnit.NANOSECONDS) :
                taskQueue.take();
            if (r != null) return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}

Saturation Strategies

When the pool cannot accept new tasks, the RejectedExecutionHandler interface provides four standard implementations:

AbortPolicy (default): Throws RejectedExecutionException

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    throw new RejectedExecutionException("Task " + r.toString() + " rejected");
}

DiscardPolicy: Silently drops the task

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    // No operation
}

DiscardOldestPolicy: Drops the eldest queued task and retries submission

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    if (!e.isShutdown()) {
        e.getQueue().poll();
        e.execute(r);
    }
}

CallerRunsPolicy: Executes the task in the caller's thread

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    if (!e.isShutdown()) {
        r.run();
    }
}

Graceful Termination

Ordered Shutdown

The shutdown() method initiates graceful termination by transitioning to SHUTDOWN state and interrupting idle workers:

public void shutdown() {
    final ReentrantLock lock = this.mainLock;
    lock.lock();
    try {
        checkShutdownAccess();
        advanceState(SHUTDOWN);
        interruptIdleWorkers();
        onShutdown();
    } finally {
        lock.unlock();
    }
    attemptTermination();
}

Immeidate Shutdown

shutdownNow() transitions to STOP state, interrupts all workers, and drains the queue:

public List<Runnable> shutdownNow() {
    List<Runnable> pendingTasks;
    final ReentrantLock lock = this.mainLock;
    lock.lock();
    try {
        checkShutdownAccess();
        advanceState(STOP);
        interruptAllWorkers();
        pendingTasks = drainQueue();
    } finally {
        lock.unlock();
    }
    attemptTermination();
    return pendingTasks;
}

Worker Cleanup and Pool Maintenance

When workers exit the processing loop, terminateWorker handles cleanup:

private void terminateWorker(Worker w, boolean abrupt) {
    if (abrupt) decrementWorkerCount();
    
    final ReentrantLock lock = this.mainLock;
    lock.lock();
    try {
        totalTasksCompleted += w.tasksProcessed;
        activeWorkers.remove(w);
    } finally {
        lock.unlock();
    }
    
    attemptTermination();
    
    int state = controlState.get();
    if (getRunState(state) < STOP) {
        if (!abrupt) {
            int min = allowCoreThreadTimeOut ? 0 : minPoolSize;
            if (min == 0 && !taskQueue.isEmpty()) min = 1;
            if (getWorkerCount(state) >= min) return;
        }
        spawnWorker(null, false);
    }
}

Queue Selection Strategies

Direct handoff (SynchronousQueue): Threads process tasks immediately or new threads spawn. Requires large maxPoolSize to avoid rejections.

Unbounded queues (LinkedBlockingQueue): Core threads process tasks; excess tasks queue indefinitely. Risk of resource exhaustion if production outpaces consumption.

Bounded queues (ArrayBlockingQueue): Balances throughput and resource constraints. When full, creates overflow threads up to maxPoolSize before rejecting.