1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

/**
 * Simplified worker thread implementation showing core mechanics.
 * Real implementations (like ThreadPoolExecutor) are more sophisticated.
 */
class WorkerThread extends Thread {
    private final BlockingQueue<Runnable> taskQueue;
    private volatile boolean running = true;
    
    public WorkerThread(BlockingQueue<Runnable> taskQueue) {
        this.taskQueue = taskQueue;
    }
    
    @Override
    public void run() {
        // This is the worker loop - the heart of the thread pool
        while (running) {
            try {
                // BLOCKING WAIT: Thread sleeps here until task available
                // This is key - threads aren't busy-waiting, they're parked
                Runnable task = taskQueue.take();  // Blocks if queue empty
                
                // Execute the task
                try {
                    task.run();
                } catch (Throwable t) {
                    // CRITICAL: Must catch all exceptions
                    // If we let exceptions propagate, the worker dies
                    handleTaskException(t);
                }
                
                // Task complete - loop back and wait for next task
                // Thread is NOT destroyed - it's REUSED for next task
                
            } catch (InterruptedException e) {
                // Pool is shutting down or thread is being terminated
                Thread.currentThread().interrupt();
                break;
            }
        }
        
        // Worker thread termination - only during pool shutdown
        cleanup();
    }
    
    public void shutdown() {
        running = false;
        this.interrupt();  // Wake from blocking take()
    }
    
    private void handleTaskException(Throwable t) {
        // Log the exception but don't let it kill the worker
        System.err.println("Task threw exception: " + t.getMessage());
        t.printStackTrace();
        // Worker continues to process next task
    }
    
    private void cleanup() {
        System.out.println("Worker thread shutting down");
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

import java.util.concurrent.*;
 
public class RejectionPolicyExamples {
    
    // AbortPolicy: Throws exception on rejection
    // Use when failures must be explicit
    ExecutorService abortPool = new ThreadPoolExecutor(
        4, 8, 60, TimeUnit.SECONDS,
        new ArrayBlockingQueue<>(100),
        new ThreadPoolExecutor.AbortPolicy()  // Default
    );
    
    // CallerRunsPolicy: Submitter executes the task
    // Provides natural backpressure
    ExecutorService callerRunsPool = new ThreadPoolExecutor(
        4, 8, 60, TimeUnit.SECONDS,
        new ArrayBlockingQueue<>(100),
        new ThreadPoolExecutor.CallerRunsPolicy()
    );
    
    // DiscardPolicy: Silently drops rejected tasks
    // Use for optional/lossy processing
    ExecutorService discardPool = new ThreadPoolExecutor(
        4, 8, 60, TimeUnit.SECONDS,
        new ArrayBlockingQueue<>(100),
        new ThreadPoolExecutor.DiscardPolicy()
    );
    
    // DiscardOldestPolicy: Drops oldest task in queue
    // Use when newer tasks are more valuable
    ExecutorService discardOldestPool = new ThreadPoolExecutor(
        4, 8, 60, TimeUnit.SECONDS,
        new ArrayBlockingQueue<>(100),
        new ThreadPoolExecutor.DiscardOldestPolicy()
    );
    
    // Custom rejection handler with logging
    ExecutorService monitoredPool = new ThreadPoolExecutor(
        4, 8, 60, TimeUnit.SECONDS,
        new ArrayBlockingQueue<>(100),
        (task, executor) -> {
            // Log the rejection for monitoring
            System.err.println("Task rejected! Queue size: " + 
                ((ThreadPoolExecutor) executor).getQueue().size());
            
            // Increment rejection counter for metrics
            rejectionCounter.increment();
            
            // Could retry with backoff, queue to fallback, etc.
            throw new RejectedExecutionException("Pool saturated");
        }
    );
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154

import java.util.concurrent.*;
import java.util.ArrayList;
import java.util.List;
 
/**
 * A simple thread pool implementation for educational purposes.
 * Demonstrates: worker threads, task queue, lifecycle management.
 * 
 * For production, use java.util.concurrent.ThreadPoolExecutor.
 */
public class SimpleThreadPool {
    private final int poolSize;
    private final BlockingQueue<Runnable> taskQueue;
    private final List<WorkerThread> workers;
    private volatile boolean isShutdown = false;
    
    public SimpleThreadPool(int poolSize, int queueCapacity) {
        this.poolSize = poolSize;
        this.taskQueue = new LinkedBlockingQueue<>(queueCapacity);
        this.workers = new ArrayList<>(poolSize);
        
        // Create and start worker threads
        for (int i = 0; i < poolSize; i++) {
            WorkerThread worker = new WorkerThread("Worker-" + i);
            workers.add(worker);
            worker.start();
        }
    }
    
    /**
     * Submit a task for execution.
     * Blocks if queue is full.
     */
    public void submit(Runnable task) throws InterruptedException {
        if (isShutdown) {
            throw new IllegalStateException("Pool is shutdown");
        }
        taskQueue.put(task);  // Blocks if queue full
    }
    
    /**
     * Try to submit a task without blocking.
     * Returns false if queue is full.
     */
    public boolean trySubmit(Runnable task) {
        if (isShutdown) {
            return false;
        }
        return taskQueue.offer(task);
    }
    
    /**
     * Graceful shutdown: stop accepting new tasks,
     * wait for existing tasks to complete.
     */
    public void shutdown() {
        isShutdown = true;
        // Workers will drain the queue and then stop
    }
    
    /**
     * Immediate shutdown: interrupt all workers.
     */
    public List<Runnable> shutdownNow() {
        isShutdown = true;
        
        // Interrupt all workers
        for (WorkerThread worker : workers) {
            worker.interrupt();
        }
        
        // Drain and return remaining tasks
        List<Runnable> remaining = new ArrayList<>();
        taskQueue.drainTo(remaining);
        return remaining;
    }
    
    /**
     * Wait for all workers to terminate.
     */
    public boolean awaitTermination(long timeout, TimeUnit unit) 
            throws InterruptedException {
        long deadline = System.nanoTime() + unit.toNanos(timeout);
        
        for (WorkerThread worker : workers) {
            long remaining = deadline - System.nanoTime();
            if (remaining <= 0) return false;
            worker.join(remaining / 1_000_000);  // Convert to millis
        }
        
        return workers.stream().noneMatch(Thread::isAlive);
    }
    
    /**
     * Worker thread that continuously processes tasks.
     */
    private class WorkerThread extends Thread {
        public WorkerThread(String name) {
            super(name);
        }
        
        @Override
        public void run() {
            while (true) {
                try {
                    // Wait for task with timeout
                    // Allows periodic checking of shutdown state
                    Runnable task = taskQueue.poll(100, TimeUnit.MILLISECONDS);
                    
                    if (task != null) {
                        try {
                            task.run();
                        } catch (Throwable t) {
                            System.err.println(getName() + " task failed: " + t);
                        }
                    } else if (isShutdown && taskQueue.isEmpty()) {
                        // Shutdown requested and no more tasks
                        break;
                    }
                    
                } catch (InterruptedException e) {
                    // Immediate shutdown requested
                    Thread.currentThread().interrupt();
                    break;
                }
            }
            System.out.println(getName() + " terminated");
        }
    }
    
    // Example usage
    public static void main(String[] args) throws Exception {
        SimpleThreadPool pool = new SimpleThreadPool(4, 100);
        
        // Submit 20 tasks
        for (int i = 0; i < 20; i++) {
            final int taskId = i;
            pool.submit(() -> {
                System.out.println(Thread.currentThread().getName() + 
                    " executing task " + taskId);
                try {
                    Thread.sleep(100);  // Simulate work
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }
        
        // Graceful shutdown
        pool.shutdown();
        pool.awaitTermination(5, TimeUnit.SECONDS);
        System.out.println("Pool terminated");
    }
}