"""Utilities for multithreaded parallel execution of tasks."""

import threading
import multiprocessing
from contextlib import contextmanager
from Queue import Queue

from .log import log_to_client

class TaskQueue(Queue, object):
    """Executable task queue used for multithreaded execution of multiple
    function calls. Concurrency is limited by `pool_size`."""
    def __init__(self, pool_size):
        super(TaskQueue, self).__init__()
        self.threads = []
        self.errors = []
        self.pool_semaphore = threading.Semaphore(pool_size)

    def enqueue_task(self, fn, *args, **kwargs):
        self.put((fn, args, kwargs))

    def _task_executor(self, fn, args, kwargs):
        try:
            fn(*args, **kwargs)
        except Exception as e:
            self.errors.append(e.message)
        finally:
            self.pool_semaphore.release()

    def execute(self):
        while not self.empty():
            self.pool_semaphore.acquire()
            fn, args, kwargs = self.get()
            thread = threading.Thread(target=self._task_executor,
                                            args=(fn, args, kwargs))
            self.threads.append(thread)
            thread.daemon = True
            thread.start()

        for thread in self.threads:
            thread.join()

        if self.errors:
            for error in self.errors:
                log_to_client(error)
            raise RuntimeError("Exceptions encountered during parallel task execution")

@contextmanager
def parallel_task_queue(pool_size=multiprocessing.cpu_count()*2):
    """Context manager for setting up a TaskQueue. Upon leaving the
    context manager, all tasks that were enqueued will be executed
    in parallel subject to `pool_size` concurrency constraints."""
    task_queue = TaskQueue(pool_size)
    yield task_queue
    task_queue.execute()

@@ -97,16 +97,44 @@
 cessing%0A
+import multiprocessing.pool%0A
 from con
@@ -482,106 +482,82 @@
 elf.
-threads = %5B%5D%0A        self.errors = %5B%5D%0A        self.pool_semaphore = threading.Semaphore(pool_size)
+pool = multiprocessing.pool.ThreadPool(pool_size)%0A        self.errors = %5B%5D
 %0A%0A  
@@ -810,67 +810,8 @@
 age)
-%0A        finally:%0A            self.pool_semaphore.release()
 %0A%0A  
@@ -879,189 +879,92 @@
     
-self.pool_semaphore.acquire()%0A            fn, args, kwargs = self.get()%0A            thread = threading.Thread(target=self._task_executor,%0A                                           
+fn, args, kwargs = self.get()%0A            self.pool.apply_async(self._task_executor,
  arg
@@ -997,109 +997,34 @@
     
-    
 self.
-threads.append(thread)%0A            thread.daemon = True%0A            thread.start
+pool.close
 ()%0A
-%0A
         
 for 
@@ -1023,54 +1023,17 @@
     
-for thread in self.threads:%0A            thread
+self.pool
 .joi

Use Python's secret ThreadPool implementation