LLD-017 · DESIGN PRACTICE

Task Scheduler / Thread Pool

Own queue admission, worker lifetime and cooperative cancellation with precise state transitions.

Local Java 21 reference verified: 26 checks. Learner drafts are not executed. Author self-review; independent review pending.

Existing introductory lesson · Version 1

Understand the problem

Design a fixed-worker scheduler for immediate and delayed tasks. Bound pending work, run user code outside scheduler ownership, expose results and distinguish graceful shutdown, immediate shutdown, cancellation request and actual local task completion.

Before you start

• Producer/consumer queues and heaps
• Thread interruption and cooperative cancellation
• Futures and task state machines
• Monotonic clocks and bounded admission

Scope and explicit decisions

• Local Java worker pool, 1..32 daemon workers and a positive pending-queue capacity. No persistence, distributed leases, automatic retries, recurring schedules or untrusted-code sandbox is implemented. Tasks are trusted Callable<String> bodies.
• The priority queue orders by due elapsed nanoseconds then submission ID. Delays are nonnegative and deadlines use checked arithmetic. A trusted injected clock must be nonnegative, nondecreasing, fast and nonthrowing.
• Workers use timed monitor waits and wake when new/earlier work arrives. Tests advance an AtomicLong clock then call wake, so correctness checks do not require arbitrary sleeps. Real termination waits use System.nanoTime, independent of the injected task clock.
• Queue capacity limits QUEUED tasks, not running workers or retained history. Total executing task count is bounded by worker count; results/records remain retained for process lifetime. There is no user fairness or priority-class policy.
• Cancel QUEUED removes it and completes its result as cancelled. Cancel RUNNING sets CANCEL_REQUESTED and interrupts the worker; a task that ignores/catches interruption may keep running. It becomes CANCELLED only when its local body returns/throws.
• The scheduler cannot retract remote effects performed by task code. A cancelled future does not prove an email/charge was undone. User callbacks and future awaits run outside the queue monitor.
• Graceful shutdown rejects new submissions and continues queued tasks, including delayed ones when due. Immediate shutdown cancels pending tasks and requests interruption of running tasks. Neither forcibly kills uncooperative user code.
• awaitTermination has a bounded real-time wait (0..60,000ms) and returns false if workers remain alive. Daemon threads do not keep a process alive; they do not make tasks durable. Fatal VM errors may terminate a worker without replacement.
• A completed task’s interruption flag is cleared before the next task runs. Ordinary task failure is stored in its future. No public raw CompletableFuture is exposed for arbitrary callbacks under scheduler locks.

Worked examples

Try before revealing

Requirements and invariants

R1 — Pending queue count never exceeds configured admission capacity. submit protects I1.

R2 — Only an owned worker can transition a queued task to running and execute it outside the queue lock. loop protects I2.

R3 — Cancellation requested is not reported as stopped until local execution ends. cancel protects I3.

R4 — Shutdown stops new admission while preserving the declared queued/running policy. shutdown protects I1.

I1: queue.size<=queueCapacity; admission closes once shutdown starts.

I2: At most workerCount task bodies execute, and a task has one current worker owner.

I3: RUNNING→CANCEL_REQUESTED may precede actual stop; only completion produces terminal CANCELLED for running work.

Derive the design

An executor needs admission and lifecycle beyond a worker loop

A single worker polling a FIFO can run tasks, but it cannot order future deadlines or protect the process against unlimited pending work. The reference adds a deadline heap, bounded admission and owned worker state. It does not expand into a durable scheduler merely because the class is named Scheduler.

Alternative: Java executor services already supply robust production building blocks. This teaching implementation exposes the mechanics; a production choice should consider mature libraries, real clock behavior, thread factories, metrics and shutdown guarantees rather than reimplement them without need.

Responsibilities and ownership

Method contracts

What, when, why and how

Each concept is attached to the requirement it protects, its implementation, an alternative and a test.

Read and run the Java reference

1import java.time.Duration;
2import java.util.*;
3import java.util.concurrent.*;
4import java.util.function.*;
5/** Owned local worker pool with a bounded delayed-ready queue. No durability or recurring jobs. */
6public final class Solution {
7    private Solution() {}
8    public enum State { QUEUED, RUNNING, CANCEL_REQUESTED, SUCCEEDED, FAILED, CANCELLED }
9    public record View(long id,State state,long dueNanos){}
10    private static final class Task {
11        final long id,due;final Callable<String> body;final CompletableFuture<String> result=new CompletableFuture<>();
12        State state=State.QUEUED;Thread worker;boolean cancel;
13        Task(long id,long due,Callable<String> body){this.id=id;this.due=due;this.body=body;}
14    }
15    public static final class Scheduler {
16        private final Object lock=new Object();
17        private final int queueCapacity;
18        private final LongSupplier clock;
19        private final Map<Long,Task> tasks=new HashMap<>();
20        private final PriorityQueue<Task> queue=new PriorityQueue<>(Comparator.comparingLong((Task t)->t.due).thenComparingLong(t->t.id));
21        private final List<Thread> workers=new ArrayList<>();
22        private long sequence,time;private boolean accepting=true,stopNow;
23        public Scheduler(int workerCount,int queueCapacity,LongSupplier clock){
24            if(workerCount<1||workerCount>32||queueCapacity<1)throw new IllegalArgumentException("configuration");this.queueCapacity=queueCapacity;this.clock=Objects.requireNonNull(clock);
25            for(int i=0;i<workerCount;i++){var worker=new Thread(this::loop,"lld-worker-"+i);worker.setDaemon(true);workers.add(worker);worker.start();}
26        }
27        private long now(){long value=clock.getAsLong();if(value<time)throw new IllegalStateException("nonmonotonic clock");time=value;return value;}
28        public long submit(Callable<String> body,long delayNanos){
29            Objects.requireNonNull(body);if(delayNanos<0)throw new IllegalArgumentException("delay");
30            synchronized(lock){if(!accepting)throw new IllegalStateException("closed");if(queue.size()==queueCapacity)throw new RejectedExecutionException("queue full");long due=Math.addExact(now(),delayNanos);var task=new Task(++sequence,due,body);tasks.put(task.id,task);queue.add(task);lock.notifyAll();return task.id;}
31        }
32        private void loop(){
33            while(true){Task task;
34                synchronized(lock){
35                    while(true){if(stopNow||(!accepting&&queue.isEmpty()))return;
36                        Task head=queue.peek();try{
37                            if(head==null){lock.wait();continue;}
38                            long remaining=head.due-now();if(remaining>0){lock.wait(remaining/1_000_000,(int)(remaining%1_000_000));continue;}
39                            task=queue.remove();task.state=State.RUNNING;task.worker=Thread.currentThread();break;
40                        }catch(InterruptedException ignored){/* Shutdown/cancel flags are authoritative. */}
41                    }
42                }
43                String value=null;Throwable failure=null;
44                try{boolean skip;synchronized(lock){skip=task.cancel;}if(!skip)value=task.body.call();}catch(Throwable error){failure=error;}
45                synchronized(lock){task.worker=null;if(task.cancel){task.state=State.CANCELLED;task.result.cancel(false);}else if(failure!=null){task.state=State.FAILED;task.result.completeExceptionally(failure);}else{task.state=State.SUCCEEDED;task.result.complete(value);}lock.notifyAll();}
46                Thread.interrupted(); // A completed task must not leak interruption into the next one.
47                if(failure instanceof VirtualMachineError fatal)throw fatal;
48            }
49        }
50        private Task task(long id){var task=tasks.get(id);if(task==null)throw new IllegalArgumentException("task");return task;}
51        public boolean cancel(long id){synchronized(lock){var task=task(id);if(task.state==State.QUEUED){queue.remove(task);task.cancel=true;task.state=State.CANCELLED;task.result.cancel(false);lock.notifyAll();return true;}if(task.state==State.RUNNING||task.state==State.CANCEL_REQUESTED){task.cancel=true;task.state=State.CANCEL_REQUESTED;task.worker.interrupt();return true;}return false;}}
52        public View view(long id){synchronized(lock){var t=task(id);return new View(t.id,t.state,t.due);}}
53        public String await(long id,Duration timeout)throws InterruptedException,ExecutionException,TimeoutException{final Task t;synchronized(lock){t=task(id);}return t.result.get(timeout.toNanos(),TimeUnit.NANOSECONDS);}
54        public void wake(){synchronized(lock){lock.notifyAll();}}
55        public void shutdown(boolean immediately){synchronized(lock){accepting=false;if(immediately){stopNow=true;for(var t:queue){t.cancel=true;t.state=State.CANCELLED;t.result.cancel(false);}queue.clear();for(var t:tasks.values())if(t.worker!=null){t.cancel=true;t.state=State.CANCEL_REQUESTED;t.worker.interrupt();}}lock.notifyAll();}}
56        public boolean awaitTermination(long millis)throws InterruptedException{if(millis<0||millis>60000)throw new IllegalArgumentException("timeout");long start=System.nanoTime(),budget=TimeUnit.MILLISECONDS.toNanos(millis);for(var worker:workers){long remaining=budget-(System.nanoTime()-start);if(remaining<=0)return workers.stream().noneMatch(Thread::isAlive);worker.join(remaining/1_000_000,(int)(remaining%1_000_000));}return workers.stream().noneMatch(Thread::isAlive);}
57    }
58}
59

Complexity and limits

• submit/poll are O(log Q) queue operations for Q pending tasks, with expected O(1) task-ID lookup. cancel of a queued Task is O(Q) due to heap identity removal.
• Shutdown-now scans retained H task records plus Q pending tasks: O(H+Q). Views are expected O(1); termination wait is bounded by supplied real-time budget, excluding caller scheduling.
• Space includes Q pending bodies, up to W running bodies and unbounded retained records/results H. Task payload bytes are not capped by count limits.
• Due scheduling and actual task completion latency depend on trusted clock, OS scheduling and user code. No real-time, fairness or distributed execution bound is claimed.

Walk through the actual reference

These walkthrough fixtures are emitted only after the Java harness verifies their operations and expected state. Concurrency tests additionally exercise competing callers.

Mistakes and counterexamples

Engineer → Senior → Staff

Return to the lesson →

Provenance and scope of evidence

Original teaching material. The local contract is implemented by Solution.java and tested by Test.java on Java 21. Distributed persistence, real provider/device adapters and deployment are not claimed.

Reference verification is not independent editorial review, production deployment verification or a claim that distributed adapters are implemented.