Thread Pool in Practice

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

import java.util.concurrent.*;
 
/**
 * ThreadPoolExecutor: The complete, configurable thread pool.
 * Understanding each parameter is essential for production use.
 */
public class ThreadPoolExecutorAnatomy {
    
    public static ThreadPoolExecutor createPool() {
        return new ThreadPoolExecutor(
            // corePoolSize: Number of threads to keep alive even when idle.
            // These threads are pre-started and ready for work.
            // Set higher for predictable latency (no thread startup delay).
            8,
            
            // maximumPoolSize: Maximum threads when queue is full.
            // Additional threads created when queue is full and more work arrives.
            // Set higher for bursty workloads that need to scale.
            32,
            
            // keepAliveTime: How long excess threads (above core) wait before dying.
            // Shorter = faster resource recovery; Longer = ready for bursts.
            60L,
            
            // unit: Time unit for keepAliveTime.
            TimeUnit.SECONDS,
            
            // workQueue: Where tasks wait when all core threads are busy.
            // Key decision: bounded (backpressure) vs unbounded (risk of OOM).
            new LinkedBlockingQueue<>(1000),
            
            // threadFactory: How to create threads.
            // Custom factory for naming, priority, daemon status, etc.
            new ThreadFactory() {
                private int counter = 0;
                @Override
                public Thread newThread(Runnable r) {
                    Thread t = new Thread(r, "worker-pool-" + counter++);
                    t.setDaemon(false);  // Keep JVM alive until pool shuts down
                    t.setPriority(Thread.NORM_PRIORITY);
                    return t;
                }
            },
            
            // handler: What to do when queue is full AND max threads reached.
            // Options: Abort, Discard, DiscardOldest, CallerRuns, or custom.
            new ThreadPoolExecutor.CallerRunsPolicy()
        );
    }
}

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

import java.util.concurrent.*;
 
/**
 * Common ThreadPoolExecutor configurations for different use cases.
 */
public class ExecutorConfigurations {
    
    private static final int CORES = Runtime.getRuntime().availableProcessors();
    
    /**
     * CPU-bound pool: fixed size, simple configuration.
     * Equivalent to Executors.newFixedThreadPool() but with bounded queue.
     */
    public static ExecutorService cpuBoundPool() {
        return new ThreadPoolExecutor(
            CORES, CORES,
            0L, TimeUnit.MILLISECONDS,
            new ArrayBlockingQueue<>(500),
            new ThreadPoolExecutor.CallerRunsPolicy()
        );
    }
    
    /**
     * I/O-bound pool: scales up under load.
     * Uses SynchronousQueue for immediate thread scaling.
     */
    public static ExecutorService ioBoundPool() {
        return new ThreadPoolExecutor(
            CORES,                  // Start with core threads
            CORES * 10,             // Scale up to 10x for I/O-heavy work
            60L, TimeUnit.SECONDS,  // Shrink back when idle
            new SynchronousQueue<>(), // No queue = immediate scaling
            new ThreadPoolExecutor.CallerRunsPolicy()
        );
    }
    
    /**
     * Elastic pool: balanced scaling with queuing.
     * Good for variable workloads with some queueing tolerance.
     */
    public static ExecutorService elasticPool() {
        return new ThreadPoolExecutor(
            CORES * 2,              // Reasonable baseline
            CORES * 8,              // Scale for spikes
            30L, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(100),  // Small queue
            new ThreadPoolExecutor.CallerRunsPolicy()
        );
    }
    
    /**
     * Single-threaded ordered pool.
     * Tasks execute in submission order, one at a time.
     * Use for: ordered processing, sequential operations.
     */
    public static ExecutorService orderedPool() {
        return new ThreadPoolExecutor(
            1, 1,
            0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<>()
        );
    }
    
    /**
     * Scheduled pool for periodic tasks.
     * Wraps ScheduledThreadPoolExecutor.
     */
    public static ScheduledExecutorService scheduledPool() {
        return new ScheduledThreadPoolExecutor(
            CORES,
            r -> {
                Thread t = new Thread(r, "scheduled-worker");
                t.setDaemon(true);
                return t;
            }
        );
    }
    
    /**
     * Production-ready pool with comprehensive configuration.
     */
    public static ThreadPoolExecutor productionPool(
            String poolName,
            int coreSize,
            int maxSize,
            int queueSize) {
        
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
            coreSize,
            maxSize,
            60L, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(queueSize),
            r -> {
                Thread t = new Thread(r, poolName + "-worker-" + System.nanoTime());
                t.setUncaughtExceptionHandler((thread, throwable) -> {
                    System.err.println("Uncaught exception in " + thread.getName());
                    throwable.printStackTrace();
                });
                return t;
            },
            (r, executor1) -> {
                // Custom rejection: log, metric, then caller-runs
                System.err.println("Task rejected! Queue size: " + 
                    executor1.getQueue().size());
                // Could increment rejection counter here
                if (!executor1.isShutdown()) {
                    r.run();  // Caller-runs fallback
                }
            }
        );
        
        // Pre-start core threads for faster first response
        executor.prestartAllCoreThreads();
        
        // Allow core threads to time out (optional, good for low-traffic periods)
        // executor.allowCoreThreadTimeOut(true);
        
        return executor;
    }
}

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
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189

from concurrent.futures import (
    ThreadPoolExecutor, 
    ProcessPoolExecutor,
    as_completed,
    wait,
    FIRST_COMPLETED,
    ALL_COMPLETED
)
import os
import time
from typing import List, Any
 
# Get CPU count for sizing
CPU_COUNT = os.cpu_count() or 1
 
 
class ThreadPoolPatterns:
    """Common patterns for using ThreadPoolExecutor in Python."""
    
    @staticmethod
    def basic_submit():
        """Basic task submission and result retrieval."""
        with ThreadPoolExecutor(max_workers=4) as executor:
            # Submit returns a Future immediately
            future = executor.submit(expensive_io_operation, "arg1", kwarg="value")
            
            # Block until result is ready
            result = future.result(timeout=30)  # With timeout
            print(f"Result: {result}")
    
    @staticmethod
    def map_pattern():
        """Process many items with the same function (like map())."""
        items = list(range(100))
        
        with ThreadPoolExecutor(max_workers=10) as executor:
            # map() returns results in order, lazily
            results = executor.map(
                process_item,  # Function
                items,         # Iterable of arguments
                timeout=60     # Total timeout
            )
            
            # Iterate to get results (blocks as needed)
            for item, result in zip(items, results):
                print(f"Item {item} -> {result}")
    
    @staticmethod
    def as_completed_pattern():
        """Process results as they complete (unordered, fastest first)."""
        urls = ["url1", "url2", "url3", "url4", "url5"]
        
        with ThreadPoolExecutor(max_workers=10) as executor:
            # Submit all tasks
            future_to_url = {
                executor.submit(fetch_url, url): url 
                for url in urls
            }
            
            # Process as they complete (fastest first)
            for future in as_completed(future_to_url, timeout=30):
                url = future_to_url[future]
                try:
                    result = future.result()
                    print(f"{url} completed: {result}")
                except Exception as e:
                    print(f"{url} failed: {e}")
    
    @staticmethod
    def wait_pattern():
        """Wait for specific completion conditions."""
        with ThreadPoolExecutor(max_workers=10) as executor:
            futures = [executor.submit(task, i) for i in range(10)]
            
            # Wait for ALL to complete
            done, not_done = wait(futures, return_when=ALL_COMPLETED)
            
            # Or wait for FIRST to complete
            done, not_done = wait(futures, return_when=FIRST_COMPLETED)
            
            # Or wait with timeout
            done, not_done = wait(futures, timeout=10)
            
            # Cancel remaining (best effort)
            for future in not_done:
                future.cancel()
 
 
class ProductionThreadPool:
    """Production-ready thread pool with error handling and metrics."""
    
    def __init__(
        self,
        name: str,
        max_workers: int,
        thread_name_prefix: str = None
    ):
        self.name = name
        self._executor = ThreadPoolExecutor(
            max_workers=max_workers,
            thread_name_prefix=thread_name_prefix or f"{name}-worker"
        )
        self._submitted = 0
        self._completed = 0
        self._failed = 0
    
    def submit(self, fn, *args, **kwargs):
        """Submit with automatic error handling and metrics."""
        self._submitted += 1
        future = self._executor.submit(fn, *args, **kwargs)
        future.add_done_callback(self._handle_completion)
        return future
    
    def _handle_completion(self, future):
        """Callback for task completion tracking."""
        try:
            # This re-raises any exception from the task
            future.result()
            self._completed += 1
        except Exception as e:
            self._failed += 1
            print(f"[{self.name}] Task failed: {e}")
    
    def stats(self) -> dict:
        """Get pool statistics."""
        return {
            "submitted": self._submitted,
            "completed": self._completed,
            "failed": self._failed,
            "pending": self._submitted - self._completed - self._failed,
        }
    
    def shutdown(self, wait: bool = True):
        """Graceful shutdown."""
        self._executor.shutdown(wait=wait)
    
    def __enter__(self):
        return self
    
    def __exit__(self, *args):
        self.shutdown()
 
 
# Example functions for the patterns above
def expensive_io_operation(arg, kwarg=None):
    time.sleep(0.1)  # Simulate I/O
    return f"processed {arg}"
 
def process_item(item):
    time.sleep(0.05)  # Simulate work
    return item * 2
 
def fetch_url(url):
    time.sleep(0.1)  # Simulate network
    return f"content of {url}"
 
def task(n):
    time.sleep(0.1)
    return n * n
 
 
# For CPU-bound work, use ProcessPoolExecutor
class CpuBoundExamples:
    """Examples using ProcessPoolExecutor for CPU-bound work."""
    
    @staticmethod
    def parallel_computation():
        """Parallel computation bypassing the GIL."""
        data_chunks = [list(range(i, i + 10000)) for i in range(0, 100000, 10000)]
        
        # ProcessPoolExecutor creates separate processes
        with ProcessPoolExecutor(max_workers=CPU_COUNT) as executor:
            results = list(executor.map(sum_of_squares, data_chunks))
        
        total = sum(results)
        print(f"Total: {total}")
 
 
def sum_of_squares(data):
    """CPU-intensive computation."""
    return sum(x * x for x in data)
 
 
if __name__ == "__main__":
    # Demo: Production pool with metrics
    with ProductionThreadPool("http-client", max_workers=10) as pool:
        futures = [pool.submit(fetch_url, f"https://example.com/{i}") for i in range(20)]
        wait([f for f in futures])
        print(f"Pool stats: {pool.stats()}")

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
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195

#pragma once
 
#include <vector>
#include <queue>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <future>
#include <atomic>
#include <memory>
#include <stdexcept>
 
/**
 * A modern C++ thread pool implementation.
 * 
 * Features:
 * - Fixed-size pool (optimal for CPU-bound work)
 * - Task queuing with futures for result retrieval
 * - Exception propagation through futures
 * - Graceful shutdown
 */
class ThreadPool {
public:
    /**
     * Create a thread pool with the specified number of threads.
     * If numThreads is 0, uses hardware_concurrency().
     */
    explicit ThreadPool(size_t numThreads = 0)
        : stop_(false)
        , activeCount_(0) {
        
        if (numThreads == 0) {
            numThreads = std::thread::hardware_concurrency();
            if (numThreads == 0) numThreads = 4;  // Fallback
        }
        
        workers_.reserve(numThreads);
        for (size_t i = 0; i < numThreads; ++i) {
            workers_.emplace_back([this] { workerLoop(); });
        }
    }
    
    // Disable copy/move
    ThreadPool(const ThreadPool&) = delete;
    ThreadPool& operator=(const ThreadPool&) = delete;
    
    /**
     * Destructor: shutdown and join all threads.
     */
    ~ThreadPool() {
        shutdown();
    }
    
    /**
     * Submit a task for execution.
     * Returns a future for the result.
     * 
     * Usage:
     *   auto future = pool.submit([](int x) { return x * 2; }, 42);
     *   int result = future.get();  // Blocks until complete
     */
    template<typename F, typename... Args>
    auto submit(F&& f, Args&&... args) 
        -> std::future<decltype(f(args...))> {
        
        using ReturnType = decltype(f(args...));
        
        // Wrap the function and its arguments in a packaged_task
        auto task = std::make_shared<std::packaged_task<ReturnType()>>(
            std::bind(std::forward<F>(f), std::forward<Args>(args)...)
        );
        
        std::future<ReturnType> result = task->get_future();
        
        {
            std::unique_lock<std::mutex> lock(queueMutex_);
            
            if (stop_) {
                throw std::runtime_error("Submit on stopped ThreadPool");
            }
            
            tasks_.emplace([task]() { (*task)(); });
        }
        
        condition_.notify_one();
        return result;
    }
    
    /**
     * Shutdown the pool.
     * If wait is true, completes all queued tasks first.
     */
    void shutdown(bool wait = true) {
        {
            std::unique_lock<std::mutex> lock(queueMutex_);
            if (stop_) return;  // Already stopped
            stop_ = true;
        }
        
        condition_.notify_all();
        
        for (std::thread& worker : workers_) {
            if (worker.joinable()) {
                worker.join();
            }
        }
    }
    
    /**
     * Get number of threads in the pool.
     */
    size_t size() const {
        return workers_.size();
    }
    
    /**
     * Get number of pending tasks.
     */
    size_t pending() const {
        std::unique_lock<std::mutex> lock(queueMutex_);
        return tasks_.size();
    }
    
    /**
     * Get number of actively executing tasks.
     */
    size_t active() const {
        return activeCount_.load();
    }
 
private:
    void workerLoop() {
        while (true) {
            std::function<void()> task;
            
            {
                std::unique_lock<std::mutex> lock(queueMutex_);
                
                condition_.wait(lock, [this] {
                    return stop_ || !tasks_.empty();
                });
                
                if (stop_ && tasks_.empty()) {
                    return;  // Shutdown and no more work
                }
                
                task = std::move(tasks_.front());
                tasks_.pop();
            }
            
            ++activeCount_;
            try {
                task();
            } catch (...) {
                // Exception is captured in the packaged_task's future
            }
            --activeCount_;
        }
    }
    
    std::vector<std::thread> workers_;
    std::queue<std::function<void()>> tasks_;
    
    mutable std::mutex queueMutex_;
    std::condition_variable condition_;
    
    std::atomic<bool> stop_;
    std::atomic<size_t> activeCount_;
};
 
 
// Example usage:
/*
int main() {
    ThreadPool pool(4);  // 4 threads
    
    // Submit tasks and get futures
    std::vector<std::future<int>> results;
    for (int i = 0; i < 10; ++i) {
        results.push_back(pool.submit([i] {
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
            return i * i;
        }));
    }
    
    // Collect results
    for (auto& future : results) {
        std::cout << future.get() << std::endl;
    }
    
    // Pool automatically shuts down in destructor
    return 0;
}
*/

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

import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
import io.micrometer.core.instrument.*;
 
/**
 * Production-ready monitored thread pool.
 * Exposes metrics via Micrometer (works with Prometheus, Datadog, etc.)
 */
public class MonitoredThreadPool extends ThreadPoolExecutor {
    
    private final Timer taskTimer;
    private final Timer waitTimer;
    private final Counter rejectionCounter;
    private final Counter exceptionCounter;
    private final String poolName;
    
    // Track submission time for wait time calculation
    private final ConcurrentHashMap<Runnable, Long> submissionTimes = 
        new ConcurrentHashMap<>();
    
    public MonitoredThreadPool(
            String name,
            int corePoolSize,
            int maximumPoolSize,
            int queueSize,
            MeterRegistry registry) {
        
        super(
            corePoolSize,
            maximumPoolSize,
            60L, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(queueSize),
            new RejectedExecutionHandler() {
                @Override
                public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
                    throw new RejectedExecutionException("Pool saturated: " + name);
                }
            }
        );
        
        this.poolName = name;
        
        // Register metrics
        this.taskTimer = Timer.builder("threadpool.task.duration")
            .tag("pool", name)
            .description("Task execution duration")
            .register(registry);
        
        this.waitTimer = Timer.builder("threadpool.task.wait")
            .tag("pool", name)
            .description("Time task waited in queue")
            .register(registry);
        
        this.rejectionCounter = Counter.builder("threadpool.rejected")
            .tag("pool", name)
            .description("Number of rejected tasks")
            .register(registry);
        
        this.exceptionCounter = Counter.builder("threadpool.exceptions")
            .tag("pool", name)
            .description("Number of task exceptions")
            .register(registry);
        
        // Gauge for current pool state
        Gauge.builder("threadpool.active", this, ThreadPoolExecutor::getActiveCount)
            .tag("pool", name)
            .description("Number of active threads")
            .register(registry);
        
        Gauge.builder("threadpool.queue.size", this, 
            e -> e.getQueue().size())
            .tag("pool", name)
            .description("Queue depth")
            .register(registry);
        
        Gauge.builder("threadpool.pool.size", this, ThreadPoolExecutor::getPoolSize)
            .tag("pool", name)
            .description("Current pool size")
            .register(registry);
    }
    
    @Override
    public void execute(Runnable command) {
        // Track submission time
        submissionTimes.put(command, System.nanoTime());
        try {
            super.execute(command);
        } catch (RejectedExecutionException e) {
            rejectionCounter.increment();
            submissionTimes.remove(command);
            throw e;
        }
    }
    
    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        // Record queue wait time
        Long submissionTime = submissionTimes.remove(r);
        if (submissionTime != null) {
            long waitNanos = System.nanoTime() - submissionTime;
            waitTimer.record(waitNanos, TimeUnit.NANOSECONDS);
        }
        super.beforeExecute(t, r);
    }
    
    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        // Record task execution time
        // Note: actual timing requires wrapping the runnable
        
        if (t != null) {
            exceptionCounter.increment();
        }
        super.afterExecute(r, t);
    }
    
    /**
     * Submit with explicit timing wrapper.
     */
    public Future<?> submitTimed(Runnable task) {
        return super.submit(() -> {
            Timer.Sample sample = Timer.start();
            try {
                task.run();
            } finally {
                sample.stop(taskTimer);
            }
        });
    }
    
    public <T> Future<T> submitTimed(Callable<T> task) {
        return super.submit(() -> {
            Timer.Sample sample = Timer.start();
            try {
                return task.call();
            } finally {
                sample.stop(taskTimer);
            }
        });
    }
}

#!/bin/bash
# Thread dump analysis for diagnosing stuck thread pools
 
# Get thread dump from running JVM
jstack <pid> > threaddump.txt
 
# Count threads by state
echo "Thread states:"
grep "java.lang.Thread.State" threaddump.txt | sort | uniq -c
 
# Find blocked threads (potential deadlock)
echo -e "
Blocked threads:"
grep -A 5 "State: BLOCKED" threaddump.txt
 
# Find threads waiting on I/O (pool sized for CPU-bound but doing I/O?)
echo -e "
Waiting on I/O:"
grep -B 5 "socketRead|socketWrite|FileInputStream|connect" threaddump.txt
 
# Find threads in your pool (adjust name filter)
echo -e "
Your pool threads:"
grep -A 10 "pool-.*-thread-" threaddump.txt
 
# For Go: use pprof
# curl http://localhost:6060/debug/pprof/goroutine > goroutines.txt
 
# For Python: use faulthandler
# python3 -c "import faulthandler; faulthandler.dump_traceback_later(30)"