{"title":"Thread Pool Pattern","description":"","content":"Creating a new thread for every task is expensive and does not scale. You pay startup cost, burn memory per thread, and under load you can end up with thousands of threads fighting for CPU, leading to context switching and unstable latency. \n\nThe **thread pool pattern** solves this by reusing a fixed set of worker threads to execute many tasks.\n\nIn this chapter, we'll explore how thread pools work internally, implement one from scratch, and understand the design decisions that make production thread pools robust.\n\n---\n\n# What is a Thread Pool?\n\n\n> **Real-World Analogy**\n>\n> Imagine a restaurant kitchen.\n>\n> Instead of hiring a new cook every time an order comes in, the restaurant keeps a fixed team of cooks (the thread pool). Orders (tasks) arrive and are placed in a queue. As soon as a cook is free, they pick up the next order, prepare it, and then go back to the queue for more.\n>\n> If too many orders arrive at once, the queue grows and customers wait longer. The restaurant can respond by adding more cooks up to a limit, or by limiting new orders (backpressure), but the key idea stays the same: **reuse a stable set of workers to handle a stream of work efficiently**.\n\n\nA thread pool works the same way. It consists of a fixed set of worker threads and a task queue. Clients submit tasks to the queue rather than creating threads directly. Workers continuously pull tasks from the queue and execute them. When no tasks are available, workers wait efficiently rather than spinning or terminating.\n\n\n```mermaid\nflowchart LR\n subgraph Clients\n C1[Client 1]:::orange\n C2[Client 2]:::orange\n C3[Client 3]:::orange\n end\n\n subgraph Pool[\"Thread Pool\"]\n direction TB\n Q[(Task Queue)]:::purple\n subgraph Workers[\"Worker Threads\"]\n W1[Worker 1]:::primary\n W2[Worker 2]:::primary\n W3[Worker 3]:::primary\n end\n end\n\n C1 --> Q\n C2 --> Q\n C3 --> Q\n Q --> W1\n Q --> W2\n Q --> W3\n\n W1 --> R1[Result]:::green\n W2 --> R2[Result]:::green\n W3 --> R3[Result]:::green\n\n classDef primary fill:#00ceff,stroke:#000,color:#000\n classDef orange fill:#ffa94d,stroke:#000,color:#000\n classDef purple fill:#9775fa,stroke:#000,color:#000\n classDef green fill:#69db7c,stroke:#000,color:#000\n```\n\n\nThe diagram shows the core structure. Clients submit tasks to a shared queue. Worker threads pull from this queue and produce results. The key insight is the decoupling: clients don't know or care which worker handles their task, and workers don't know or care where tasks come from. This separation enables efficient resource usage and clean architecture.\n\nThe key properties of a thread pool:\n\n- **Fixed resource usage:** Thread count is bounded, preventing resource exhaustion\n- **Task reuse:** Threads are reused across many tasks, amortizing creation cost\n- **Decoupling:** Task submission is separate from task execution\n- **Backpressure:** Queue provides natural flow control when workers are overwhelmed\n\n### Benefits of Thread Pools\n\nCreating a thread is expensive. On Linux, spawning a thread takes roughly 10-30 microseconds and allocates 1-8 MB of stack space. For a server handling 10,000 requests per second, that's 100-300 milliseconds of pure thread creation overhead every second, plus gigabytes of memory if requests overlap.\n\nBeyond raw cost, there's the thundering herd problem. Under load spikes, creating thousands of threads simultaneously overwhelms the scheduler. Context switching becomes the dominant activity. Your CPU spends more time switching between threads than doing actual work.\n\n#### Real Systems Using Thread Pools\n\n\n| System | How It Uses Thread Pools |\n|--------|-------------------------|\n| Java ExecutorService | The standard way to manage concurrency in Java applications. Spring Boot, Tomcat, and most JVM services use this. |\n| Python ThreadPoolExecutor | Part of `concurrent.futures`, used for I/O-bound parallel tasks despite the GIL. |\n| Nginx | Worker processes (similar concept) with fixed count, each handling thousands of connections via event loops. |\n| Node.js libuv | Thread pool for file I/O and DNS lookups, since these can't be done asynchronously at the OS level. |\n| Database Connection Pools | Same pattern applied to database connections: HikariCP, c3p0, pgbouncer. |\n\n\n---\n\n# Core Components\n\nA thread pool consists of four essential components. Understanding each one is crucial for both using existing implementations and designing custom solutions.\n\n### Task Queue\n\nThe task queue is where submitted tasks wait until a worker is available. This is the buffer between producers (clients submitting work) and consumers (worker threads).\n\n#### **Key Responsibilities**\n\n- Store pending tasks in submission order (FIFO typically)\n- Block workers when empty (no busy waiting)\n- Block or reject producers when full (backpressure)\n- Support concurrent access from multiple threads\n\n#### **Bounded vs. Unbounded Queue**\n\n\n| Type | Pros | Cons |\n|------|------|------|\n| Bounded | Memory usage controlled, provides backpressure, fails fast under overload | Requires rejection handling, may block producers |\n| Unbounded | Simple, never rejects tasks | Memory can grow without limit, hides overload, can cause OOM |\n\n\nProduction systems almost always use bounded queues. An unbounded queue just moves the failure from \"rejected task\" to \"out of memory exception\" with no warning.\n\n### Worker Threads\n\nWorkers are the actual threads that execute tasks. They run a simple loop: take a task from the queue, execute it, repeat.\n\n#### **Key Responsibilities**\n\n- Wait efficiently when no tasks are available\n- Execute tasks to completion\n- Handle exceptions without dying\n- Respond to shutdown signals\n\n#### **Design Decisions**\n\n- Should workers catch exceptions from tasks, or let them propagate?\n- Should workers timeout and die if idle too long (dynamic sizing)?\n- How should workers signal completion or failure?\n\n\n```mermaid\nstateDiagram-v2\n [*] --> Idle: Created\n\n Idle --> Waiting: No tasks\n Waiting --> Executing: Task available\n Executing --> Idle: Task complete\n Executing --> Idle: Task failed
(catch exception)\n\n Idle --> Terminated: Shutdown signal\n Waiting --> Terminated: Shutdown signal\n\n classDef idle fill:#69db7c,stroke:#000,color:#000\n classDef waiting fill:#ffa94d,stroke:#000,color:#000\n classDef executing fill:#00ceff,stroke:#000,color:#000\n classDef terminated fill:#ff8787,stroke:#000,color:#000\n\n class Idle idle\n class Waiting waiting\n class Executing executing\n class Terminated terminated\n```\n\n\n### Thread Manager\n\nThe thread manager (often called the pool itself) coordinates workers and handles lifecycle concerns.\n\n#### **Key Responsibilities**\n\n- Create and start worker threads at initialization\n- Track which workers are alive and which have died\n- Handle graceful shutdown (finish current tasks) vs. immediate shutdown (interrupt everything)\n- Potentially resize the pool dynamically\n\n#### **Shutdown Strategies**\n\n\n| Strategy | Behavior |\n|----------|----------|\n| Graceful (`shutdown()`) | Stop accepting new tasks, let queued tasks complete, then terminate workers |\n| Immediate (`shutdownNow()`) | Stop accepting tasks, interrupt running tasks, discard queued tasks, terminate workers |\n\n\n### Rejection Handler\n\nWhen the queue is full and all workers are busy, new tasks must be handled somehow. The rejection handler defines this policy.\n\n#### **Common Rejection Policies**\n\n\n| Policy | Behavior | Use Case |\n|--------|----------|----------|\n| Abort | Throw exception | Fail fast, let caller handle |\n| Discard | Silently drop task | Fire-and-forget tasks, logging |\n| Discard Oldest | Drop oldest queued task, add new | Fresh data more important than old |\n| Caller Runs | Execute in submitter's thread | Natural backpressure, slows down producer |\n\n\n\n```mermaid\nflowchart TD\n NEW[New Task Submitted]:::orange --> CHECK{Queue
Full?}\n CHECK -->|No| QUEUE[Add to Queue]:::green\n CHECK -->|Yes| POLICY{Rejection
Policy}\n\n POLICY -->|Abort| THROW[Throw Exception]:::red\n POLICY -->|Discard| DROP[Drop Task]:::red\n POLICY -->|Discard Oldest| REMOVE[Remove Oldest
Add New]:::yellow\n POLICY -->|Caller Runs| CALLER[Execute in
Caller Thread]:::purple\n\n classDef orange fill:#ffa94d,stroke:#000,color:#000\n classDef green fill:#69db7c,stroke:#000,color:#000\n classDef red fill:#ff8787,stroke:#000,color:#000\n classDef yellow fill:#ffd43b,stroke:#000,color:#000\n classDef purple fill:#9775fa,stroke:#000,color:#000\n```\n\n\nThe Caller Runs policy is particularly clever. When the pool is overwhelmed, the submitting thread is forced to do the work itself. This automatically slows down the producer, creating natural backpressure without explicit coordination.\n\n---\n\n# How It Works\n\nLet's trace through the lifecycle of a task from submission to completion.\n\n#### **Step 1: Task Submission**\n\nA client calls `submit(task)`. The pool checks if the queue has space. If yes, the task is added to the queue and the method returns immediately (usually with a Future for the result).\n\n#### **Step 2: Queue Wait**\n\nThe task sits in the queue. Meanwhile, worker threads are either executing other tasks or waiting for work.\n\n#### **Step 3: Worker Dequeues Task**\n\nWhen a worker finishes its current task (or was already idle), it calls `queue.take()`. This is a blocking call. If the queue is empty, the worker sleeps until a task arrives. When our task reaches the front and a worker is available, the worker wakes up with the task in hand.\n\n#### **Step 4: Task Execution**\n\nThe worker calls `task.run()`. This is where the actual work happens. The worker thread's stack, CPU time, and context are all devoted to this task.\n\n#### **Step 5: Completion**\n\nThe task finishes (successfully or with an exception). If a Future was returned at submission, it's now completed. The worker loops back to Step 3, ready for the next task.\n\n#### **Step 6: Result Retrieval**\n\nThe client, whenever it's ready, calls `future.get()` to retrieve the result. If the task isn't done yet, this blocks. If it completed with an exception, the exception is rethrown.\n\n\n```mermaid\nsequenceDiagram\n participant Client\n participant Pool as Thread Pool\n participant Queue as Task Queue\n participant Worker\n participant Task\n\n Client->>Pool: submit(task)\n Pool->>Queue: offer(task)\n Pool-->>Client: Future\n\n Note over Queue: Task waits...\n\n Worker->>Queue: take()\n Queue-->>Worker: task\n\n Worker->>Task: run()\n activate Task\n Task-->>Worker: result/exception\n deactivate Task\n\n Worker->>Pool: complete(future, result)\n\n Client->>Pool: future.get()\n Pool-->>Client: result\n```\n\n\n#### **Example Trace**\n\n\n```plaintext\nInitial State:\n Queue: []\n Workers: [W1: idle, W2: idle, W3: executing Task-A]\n\nt=0: Client submits Task-B\n Queue: [Task-B]\n W1 wakes up, takes Task-B\n Queue: []\n Workers: [W1: executing Task-B, W2: idle, W3: executing Task-A]\n\nt=5ms: Client submits Task-C and Task-D\n Queue: [Task-C, Task-D]\n W2 wakes up, takes Task-C\n Queue: [Task-D]\n Workers: [W1: executing Task-B, W2: executing Task-C, W3: executing Task-A]\n\nt=10ms: W3 completes Task-A\n W3 takes Task-D\n Queue: []\n Workers: [W1: executing Task-B, W2: executing Task-C, W3: executing Task-D]\n\nt=20ms: All tasks complete\n Workers: [W1: idle, W2: idle, W3: idle]\n All waiting for next task...\n```\n\n\n---\n\n# Implementation\n\nLet's implement a basic thread pool from scratch. This helps you understand what's happening inside `ExecutorService` or `ThreadPoolExecutor`.\n\n\n#### Java\n\n\n```java\nimport java.util.concurrent.*;\nimport java.util.concurrent.atomic.AtomicBoolean;\n\npublic class SimpleThreadPool {\n private final BlockingQueue taskQueue;\n private final Thread[] workers;\n private final AtomicBoolean isShutdown = new AtomicBoolean(false);\n\n public SimpleThreadPool(int numThreads, int queueCapacity) {\n this.taskQueue = new LinkedBlockingQueue<>(queueCapacity);\n this.workers = new Thread[numThreads];\n\n // Create and start worker threads\n for (int i = 0; i < numThreads; i++) {\n workers[i] = new Thread(this::workerLoop, \"Worker-\" + i);\n workers[i].start();\n }\n }\n\n private void workerLoop() {\n while (!isShutdown.get()) {\n try {\n // Block until a task is available\n Runnable task = taskQueue.take();\n try {\n task.run();\n } catch (Exception e) {\n // Log but don't die - worker continues\n System.err.println(\"Task failed: \" + e.getMessage());\n }\n } catch (InterruptedException e) {\n // Shutdown signal received\n Thread.currentThread().interrupt();\n break;\n }\n }\n }\n\n public void submit(Runnable task) throws RejectedExecutionException {\n if (isShutdown.get()) {\n throw new RejectedExecutionException(\"Pool is shutdown\");\n }\n if (!taskQueue.offer(task)) {\n throw new RejectedExecutionException(\"Queue is full\");\n }\n }\n\n public void shutdown() {\n isShutdown.set(true);\n for (Thread worker : workers) {\n worker.interrupt();\n }\n }\n\n public void awaitTermination() throws InterruptedException {\n for (Thread worker : workers) {\n worker.join();\n }\n }\n}\n```\n\n\n**Key Points:**\n\n- Line 3: `BlockingQueue` handles all the thread-safe waiting and signaling\n- Line 16-18: `take()` blocks efficiently until a task is available\n- Line 19-23: Task exceptions are caught so the worker survives\n- Line 31-34: `offer()` returns false if queue is full, triggering rejection\n\n\n### Using the Standard Library\n\nIn practice, you should use the standard library implementations. They handle edge cases we've glossed over.\n\n\n```java\nimport java.util.concurrent.*;\n\npublic class StandardPoolExample {\n public static void main(String[] args) throws Exception {\n // Fixed thread pool with bounded queue\n ThreadPoolExecutor executor = new ThreadPoolExecutor(\n 4, // core pool size\n 4, // max pool size\n 0L, TimeUnit.MILLISECONDS, // keep-alive (unused with fixed size)\n new ArrayBlockingQueue<>(100), // bounded queue\n new ThreadPoolExecutor.CallerRunsPolicy() // rejection policy\n );\n\n // Submit tasks\n Future future = executor.submit(() -> {\n Thread.sleep(100);\n return 42;\n });\n\n System.out.println(\"Result: \" + future.get());\n\n // Graceful shutdown\n executor.shutdown();\n executor.awaitTermination(10, TimeUnit.SECONDS);\n }\n}\n```\n\n```python\nfrom concurrent.futures import ThreadPoolExecutor, as_completed\nimport time\n\ndef main():\n # ThreadPoolExecutor with max workers\n with ThreadPoolExecutor(max_workers=4) as executor:\n # Submit tasks\n futures = []\n for i in range(10):\n future = executor.submit(process_task, i)\n futures.append(future)\n\n # Process results as they complete\n for future in as_completed(futures):\n try:\n result = future.result()\n print(f\"Result: {result}\")\n except Exception as e:\n print(f\"Task failed: {e}\")\n\n # Executor automatically shuts down when exiting 'with' block\n\ndef process_task(task_id):\n time.sleep(0.1)\n return task_id * 2\n\nif __name__ == \"__main__\":\n main()\n```\n\n```cpp\n// C++17 doesn't have a standard thread pool, but you can use:\n// - Boost.Asio's thread pool\n// - Intel TBB's task scheduler\n// - Roll your own (like above) or use a header-only library\n\n// Example with a simple wrapper around std::async\n#include \n#include \n\nint main() {\n std::vector> futures;\n\n // Launch async tasks (uses internal thread pool on most implementations)\n for (int i = 0; i < 10; ++i) {\n futures.push_back(std::async(std::launch::async, [i] {\n std::this_thread::sleep_for(std::chrono::milliseconds(100));\n return i * 2;\n }));\n }\n\n // Collect results\n for (auto& future : futures) {\n std::cout << \"Result: \" << future.get() << std::endl;\n }\n\n return 0;\n}\n```\n\n```csharp\nusing System;\nusing System.Collections.Generic;\nusing System.Threading;\nusing System.Threading.Tasks;\n\npublic class StandardPoolExample\n{\n public static async Task Main(string[] args)\n {\n // Configure ThreadPool limits (optional)\n ThreadPool.SetMinThreads(4, 4);\n ThreadPool.SetMaxThreads(100, 100);\n\n // Option 1: Using Task.Run (recommended for most scenarios)\n var tasks = new List>();\n\n for (int i = 0; i < 10; i++)\n {\n int taskId = i; // Capture for closure\n var task = Task.Run(async () =>\n {\n await Task.Delay(100);\n return taskId * 2;\n });\n tasks.Add(task);\n }\n\n // Await all tasks and process results\n var results = await Task.WhenAll(tasks);\n foreach (var result in results)\n {\n Console.WriteLine($\"Result: {result}\");\n }\n\n // Option 2: Using ThreadPool.QueueUserWorkItem directly\n using var countdown = new CountdownEvent(3);\n\n for (int i = 0; i < 3; i++)\n {\n int workItem = i;\n ThreadPool.QueueUserWorkItem(_ =>\n {\n Console.WriteLine($\"Processing work item {workItem} on thread {Environment.CurrentManagedThreadId}\");\n countdown.Signal();\n });\n }\n\n countdown.Wait(); // Wait for all work items to complete\n Console.WriteLine(\"All work items completed\");\n }\n}\n```\n\n```go\npackage main\n\nimport (\n\t\"context\"\n\t\"fmt\"\n\t\"sync\"\n\t\"time\"\n)\n\n// Job represents work to be done\ntype Job struct {\n\tID int\n\tProcess func() (int, error)\n}\n\n// Result holds the outcome of a job\ntype Result struct {\n\tJobID int\n\tValue int\n\tErr error\n}\n\n// WorkerPool manages a pool of workers processing jobs\ntype WorkerPool struct {\n\tnumWorkers int\n\tjobs chan Job\n\tresults chan Result\n\twg sync.WaitGroup\n}\n\nfunc NewWorkerPool(numWorkers, jobBuffer int) *WorkerPool {\n\treturn &WorkerPool{\n\t\tnumWorkers: numWorkers,\n\t\tjobs: make(chan Job, jobBuffer),\n\t\tresults: make(chan Result, jobBuffer),\n\t}\n}\n\nfunc (wp *WorkerPool) Start(ctx context.Context) {\n\tfor i := 0; i < wp.numWorkers; i++ {\n\t\twp.wg.Add(1)\n\t\tgo wp.worker(ctx, i)\n\t}\n}\n\nfunc (wp *WorkerPool) worker(ctx context.Context, id int) {\n\tdefer wp.wg.Done()\n\n\tfor {\n\t\tselect {\n\t\tcase <-ctx.Done():\n\t\t\treturn\n\t\tcase job, ok := <-wp.jobs:\n\t\t\tif !ok {\n\t\t\t\treturn\n\t\t\t}\n\t\t\tvalue, err := job.Process()\n\t\t\twp.results <- Result{JobID: job.ID, Value: value, Err: err}\n\t\t}\n\t}\n}\n\nfunc (wp *WorkerPool) Submit(job Job) {\n\twp.jobs <- job\n}\n\nfunc (wp *WorkerPool) Results() <-chan Result {\n\treturn wp.results\n}\n\nfunc (wp *WorkerPool) Shutdown() {\n\tclose(wp.jobs)\n\twp.wg.Wait()\n\tclose(wp.results)\n}\n\nfunc main() {\n\tctx, cancel := context.WithCancel(context.Background())\n\tdefer cancel()\n\n\tpool := NewWorkerPool(4, 100)\n\tpool.Start(ctx)\n\n\t// Collect results in background\n\tvar wg sync.WaitGroup\n\twg.Add(1)\n\tgo func() {\n\t\tdefer wg.Done()\n\t\tfor result := range pool.Results() {\n\t\t\tif result.Err != nil {\n\t\t\t\tfmt.Printf(\"Job %d failed: %v\\n\", result.JobID, result.Err)\n\t\t\t} else {\n\t\t\t\tfmt.Printf(\"Job %d result: %d\\n\", result.JobID, result.Value)\n\t\t\t}\n\t\t}\n\t}()\n\n\t// Submit jobs\n\tfor i := 0; i < 10; i++ {\n\t\tjobID := i\n\t\tpool.Submit(Job{\n\t\t\tID: jobID,\n\t\t\tProcess: func() (int, error) {\n\t\t\t\ttime.Sleep(100 * time.Millisecond)\n\t\t\t\treturn jobID * 2, nil\n\t\t\t},\n\t\t})\n\t}\n\n\tpool.Shutdown()\n\twg.Wait()\n}\n```\n\n\n---\n\n# Practical Example\n\n### Scenario: Parallel Image Processor\n\nYou're building an image processing service. Users upload images, and you need to generate thumbnails, apply filters, and extract metadata. Each operation is CPU-intensive and independent.\n\n**Requirements:**\n\n- Process multiple images concurrently\n- Limit resource usage (don't spawn unlimited threads)\n- Handle failures gracefully (one bad image shouldn't crash everything)\n- Provide progress feedback\n\n\n```mermaid\nflowchart LR\n subgraph Input\n I1[Image 1]:::orange\n I2[Image 2]:::orange\n I3[Image 3]:::orange\n end\n\n subgraph Pool[\"Image Processing Pool (4 workers)\"]\n Q[(Task Queue)]:::purple\n W1[Worker 1]:::primary\n W2[Worker 2]:::primary\n W3[Worker 3]:::primary\n W4[Worker 4]:::primary\n end\n\n subgraph Tasks[\"Per-Image Tasks\"]\n T[Thumbnail]:::secondary\n F[Filter]:::secondary\n M[Metadata]:::secondary\n end\n\n I1 --> Q\n I2 --> Q\n I3 --> Q\n\n Q --> W1\n Q --> W2\n Q --> W3\n Q --> W4\n\n W1 --> T\n W2 --> F\n W3 --> M\n\n classDef primary fill:#00ceff,stroke:#000,color:#000\n classDef secondary fill:#38d9a9,stroke:#000,color:#000\n classDef orange fill:#ffa94d,stroke:#000,color:#000\n classDef purple fill:#9775fa,stroke:#000,color:#000\n```\n\n\n### Implementation\n\n\n```java\nimport java.util.concurrent.*;\nimport java.util.List;\nimport java.util.ArrayList;\n\npublic class ImageProcessor {\n private final ExecutorService pool;\n\n public ImageProcessor(int workers) {\n this.pool = new ThreadPoolExecutor(\n workers, workers,\n 0L, TimeUnit.MILLISECONDS,\n new ArrayBlockingQueue<>(1000),\n new ThreadPoolExecutor.CallerRunsPolicy()\n );\n }\n\n public List processImages(List imagePaths) {\n List> futures = new ArrayList<>();\n\n for (String path : imagePaths) {\n Future future = pool.submit(() -> {\n try {\n byte[] thumbnail = generateThumbnail(path);\n byte[] filtered = applyFilters(path);\n String metadata = extractMetadata(path);\n return new ProcessingResult(path, thumbnail, filtered, metadata, null);\n } catch (Exception e) {\n return new ProcessingResult(path, null, null, null, e.getMessage());\n }\n });\n futures.add(future);\n }\n\n List results = new ArrayList<>();\n for (Future future : futures) {\n try {\n results.add(future.get(30, TimeUnit.SECONDS));\n } catch (TimeoutException e) {\n results.add(new ProcessingResult(\"unknown\", null, null, null, \"Timeout\"));\n } catch (Exception e) {\n results.add(new ProcessingResult(\"unknown\", null, null, null, e.getMessage()));\n }\n }\n return results;\n }\n\n public void shutdown() {\n pool.shutdown();\n }\n\n // Placeholder methods\n private byte[] generateThumbnail(String path) { return new byte[0]; }\n private byte[] applyFilters(String path) { return new byte[0]; }\n private String extractMetadata(String path) { return \"metadata\"; }\n}\n\nrecord ProcessingResult(\n String path,\n byte[] thumbnail,\n byte[] filtered,\n String metadata,\n String error\n) {}\n```\n\n```python\nfrom concurrent.futures import ThreadPoolExecutor, as_completed\nfrom dataclasses import dataclass\nfrom typing import Optional\nimport time\n\n@dataclass\nclass ProcessingResult:\n path: str\n thumbnail: Optional[bytes]\n filtered: Optional[bytes]\n metadata: Optional[str]\n error: Optional[str]\n\nclass ImageProcessor:\n def __init__(self, workers: int):\n self.pool = ThreadPoolExecutor(max_workers=workers)\n\n def process_images(self, image_paths: list[str]) -> list[ProcessingResult]:\n futures = {}\n for path in image_paths:\n future = self.pool.submit(self._process_single, path)\n futures[future] = path\n\n results = []\n for future in as_completed(futures, timeout=60):\n path = futures[future]\n try:\n result = future.result(timeout=30)\n results.append(result)\n except TimeoutError:\n results.append(ProcessingResult(path, None, None, None, \"Timeout\"))\n except Exception as e:\n results.append(ProcessingResult(path, None, None, None, str(e)))\n\n return results\n\n def _process_single(self, path: str) -> ProcessingResult:\n try:\n thumbnail = self._generate_thumbnail(path)\n filtered = self._apply_filters(path)\n metadata = self._extract_metadata(path)\n return ProcessingResult(path, thumbnail, filtered, metadata, None)\n except Exception as e:\n return ProcessingResult(path, None, None, None, str(e))\n\n def shutdown(self):\n self.pool.shutdown(wait=True)\n\n # Placeholder methods\n def _generate_thumbnail(self, path: str) -> bytes:\n time.sleep(0.1) # Simulate work\n return b\"\"\n\n def _apply_filters(self, path: str) -> bytes:\n return b\"\"\n\n def _extract_metadata(self, path: str) -> str:\n return \"metadata\"\n\n# Usage\nif __name__ == \"__main__\":\n processor = ImageProcessor(workers=4)\n results = processor.process_images([\"img1.jpg\", \"img2.jpg\", \"img3.jpg\"])\n for r in results:\n status = \"OK\" if r.error is None else f\"FAILED: {r.error}\"\n print(f\"{r.path}: {status}\")\n processor.shutdown()\n```\n\n```csharp\nusing System;\nusing System.Collections.Concurrent;\nusing System.Collections.Generic;\nusing System.Threading;\nusing System.Threading.Tasks;\n\npublic class ProcessingResult\n{\n public string Path { get; }\n public byte[]? Thumbnail { get; }\n public byte[]? Filtered { get; }\n public string? Metadata { get; }\n public string? Error { get; }\n\n public ProcessingResult(string path, byte[]? thumbnail, byte[]? filtered,\n string? metadata, string? error)\n {\n Path = path;\n Thumbnail = thumbnail;\n Filtered = filtered;\n Metadata = metadata;\n Error = error;\n }\n}\n\npublic class ImageProcessor : IDisposable\n{\n private readonly SemaphoreSlim _semaphore;\n private readonly int _maxWorkers;\n\n public ImageProcessor(int workers)\n {\n _maxWorkers = workers;\n _semaphore = new SemaphoreSlim(workers, workers);\n }\n\n public async Task> ProcessImagesAsync(\n List imagePaths, CancellationToken cancellationToken = default)\n {\n var tasks = new List>();\n\n foreach (var path in imagePaths)\n {\n var task = ProcessWithThrottlingAsync(path, cancellationToken);\n tasks.Add(task);\n }\n\n var results = new List();\n foreach (var task in tasks)\n {\n try\n {\n var result = await task.WaitAsync(TimeSpan.FromSeconds(30), cancellationToken);\n results.Add(result);\n }\n catch (TimeoutException)\n {\n results.Add(new ProcessingResult(\"unknown\", null, null, null, \"Timeout\"));\n }\n catch (Exception e)\n {\n results.Add(new ProcessingResult(\"unknown\", null, null, null, e.Message));\n }\n }\n\n return results;\n }\n\n private async Task ProcessWithThrottlingAsync(\n string path, CancellationToken cancellationToken)\n {\n await _semaphore.WaitAsync(cancellationToken);\n try\n {\n return await Task.Run(() => ProcessSingle(path), cancellationToken);\n }\n finally\n {\n _semaphore.Release();\n }\n }\n\n private ProcessingResult ProcessSingle(string path)\n {\n try\n {\n var thumbnail = GenerateThumbnail(path);\n var filtered = ApplyFilters(path);\n var metadata = ExtractMetadata(path);\n return new ProcessingResult(path, thumbnail, filtered, metadata, null);\n }\n catch (Exception e)\n {\n return new ProcessingResult(path, null, null, null, e.Message);\n }\n }\n\n // Placeholder methods\n private byte[] GenerateThumbnail(string path)\n {\n Thread.Sleep(100); // Simulate work\n return Array.Empty();\n }\n\n private byte[] ApplyFilters(string path) => Array.Empty();\n\n private string ExtractMetadata(string path) => \"metadata\";\n\n public void Dispose()\n {\n _semaphore.Dispose();\n }\n}\n\n// Usage\npublic class Program\n{\n public static async Task Main(string[] args)\n {\n using var processor = new ImageProcessor(workers: 4);\n var results = await processor.ProcessImagesAsync(\n new List { \"img1.jpg\", \"img2.jpg\", \"img3.jpg\" });\n\n foreach (var r in results)\n {\n var status = r.Error == null ? \"OK\" : $\"FAILED: {r.Error}\";\n Console.WriteLine($\"{r.Path}: {status}\");\n }\n }\n}\n```\n\n```go\npackage main\n\nimport (\n\t\"context\"\n\t\"fmt\"\n\t\"sync\"\n\t\"time\"\n)\n\ntype ProcessingResult struct {\n\tPath string\n\tThumbnail []byte\n\tFiltered []byte\n\tMetadata string\n\tError error\n}\n\ntype ImageProcessor struct {\n\tworkers int\n}\n\nfunc NewImageProcessor(workers int) *ImageProcessor {\n\treturn &ImageProcessor{workers: workers}\n}\n\nfunc (p *ImageProcessor) ProcessImages(ctx context.Context, imagePaths []string) []ProcessingResult {\n\tjobs := make(chan string, len(imagePaths))\n\tresults := make(chan ProcessingResult, len(imagePaths))\n\n\t// Start workers\n\tvar wg sync.WaitGroup\n\tfor i := 0; i < p.workers; i++ {\n\t\twg.Add(1)\n\t\tgo func() {\n\t\t\tdefer wg.Done()\n\t\t\tfor {\n\t\t\t\tselect {\n\t\t\t\tcase <-ctx.Done():\n\t\t\t\t\treturn\n\t\t\t\tcase path, ok := <-jobs:\n\t\t\t\t\tif !ok {\n\t\t\t\t\t\treturn\n\t\t\t\t\t}\n\t\t\t\t\tresults <- p.processSingle(ctx, path)\n\t\t\t\t}\n\t\t\t}\n\t\t}()\n\t}\n\n\t// Submit all jobs\n\tfor _, path := range imagePaths {\n\t\tjobs <- path\n\t}\n\tclose(jobs)\n\n\t// Wait for workers and close results\n\tgo func() {\n\t\twg.Wait()\n\t\tclose(results)\n\t}()\n\n\t// Collect results with timeout\n\tvar output []ProcessingResult\n\tfor result := range results {\n\t\toutput = append(output, result)\n\t}\n\n\treturn output\n}\n\nfunc (p *ImageProcessor) processSingle(ctx context.Context, path string) ProcessingResult {\n\t// Create timeout context for this task\n\ttaskCtx, cancel := context.WithTimeout(ctx, 30*time.Second)\n\tdefer cancel()\n\n\tresultCh := make(chan ProcessingResult, 1)\n\n\tgo func() {\n\t\tthumbnail := p.generateThumbnail(path)\n\t\tfiltered := p.applyFilters(path)\n\t\tmetadata := p.extractMetadata(path)\n\t\tresultCh <- ProcessingResult{\n\t\t\tPath: path,\n\t\t\tThumbnail: thumbnail,\n\t\t\tFiltered: filtered,\n\t\t\tMetadata: metadata,\n\t\t}\n\t}()\n\n\tselect {\n\tcase <-taskCtx.Done():\n\t\treturn ProcessingResult{Path: path, Error: taskCtx.Err()}\n\tcase result := <-resultCh:\n\t\treturn result\n\t}\n}\n\n// Placeholder methods\nfunc (p *ImageProcessor) generateThumbnail(path string) []byte {\n\ttime.Sleep(100 * time.Millisecond) // Simulate work\n\treturn []byte{}\n}\n\nfunc (p *ImageProcessor) applyFilters(path string) []byte {\n\treturn []byte{}\n}\n\nfunc (p *ImageProcessor) extractMetadata(path string) string {\n\treturn \"metadata\"\n}\n\nfunc main() {\n\tctx := context.Background()\n\tprocessor := NewImageProcessor(4)\n\n\tresults := processor.ProcessImages(ctx, []string{\"img1.jpg\", \"img2.jpg\", \"img3.jpg\"})\n\n\tfor _, r := range results {\n\t\tstatus := \"OK\"\n\t\tif r.Error != nil {\n\t\t\tstatus = fmt.Sprintf(\"FAILED: %v\", r.Error)\n\t\t}\n\t\tfmt.Printf(\"%s: %s\\n\", r.Path, status)\n\t}\n}\n```\n\n\n---\n\n# When to Use / When Not to Use\n\n\n| Use When | Avoid When |\n|----------|------------|\n| You have many independent tasks to execute | Tasks are heavily interdependent |\n| Task creation overhead is significant | You only have a few long-running tasks |\n| You need to limit concurrent resource usage | Each task needs dedicated resources (e.g., GPU) |\n| Tasks are short-lived relative to thread creation | Tasks run for the entire application lifetime |\n| You want to decouple task submission from execution | You need fine-grained control over each thread |\n\n\n#### **Consider Instead:**\n\n- **Single thread:** If tasks must be serialized anyway\n- **Fork/Join:** For recursive divide-and-conquer algorithms\n- **Async/Await:** For I/O-bound work in languages with good async support\n- **Actor model:** When tasks need to maintain state and communicate\n\n---\n\n# Quiz","pageType":"concurrency-interview"}

Get Premium