Threading pool similar to the multiprocessing Pool?

In Python 3 you can use concurrent.futures.ThreadPoolExecutor, i.e.:

executor = ThreadPoolExecutor(max_workers=10) a = executor.submit(my_function) 

See the docs for more info and examples.

Yes, and it seems to have (more or less) the same API.

import multiprocessing def worker(lnk): .... def start_process(): ..... .... if(PROCESS): pool = multiprocessing.Pool(processes=POOL_SIZE, initializer=start_process) else: pool = multiprocessing.pool.ThreadPool(processes=POOL_SIZE, initializer=start_process) pool.map(worker, inputs) .... 

For something very simple and lightweight (slightly modified from here):

from Queue import Queue from threading import Thread class Worker(Thread): """Thread executing tasks from a given tasks queue""" def __init__(self, tasks): Thread.__init__(self) self.tasks = tasks self.daemon = True self.start() def run(self): while True: func, args, kargs = self.tasks.get() try: func(*args, **kargs) except Exception, e: print e finally: self.tasks.task_done() class ThreadPool: """Pool of threads consuming tasks from a queue""" def __init__(self, num_threads): self.tasks = Queue(num_threads) for _ in range(num_threads): Worker(self.tasks) def add_task(self, func, *args, **kargs): """Add a task to the queue""" self.tasks.put((func, args, kargs)) def wait_completion(self): """Wait for completion of all the tasks in the queue""" self.tasks.join() if __name__ == '__main__': from random import randrange from time import sleep delays = [randrange(1, 10) for i in range(100)] def wait_delay(d): print 'sleeping for (%d)sec' % d sleep(d) pool = ThreadPool(20) for i, d in enumerate(delays): pool.add_task(wait_delay, d) pool.wait_completion() 

To support callbacks on task completion you can just add the callback to the task tuple.

Hi to use the thread pool in Python you can use this library :

from multiprocessing.dummy import Pool as ThreadPool 

and then for use, this library do like that :

pool = ThreadPool(threads) results = pool.map(service, tasks) pool.close() pool.join() return results 

The threads are the number of threads that you want and tasks are a list of task that most map to the service.

Yes, there is a threading pool similar to the multiprocessing Pool, however, it is hidden somewhat and not properly documented. You can import it by following way:-

from multiprocessing.pool import ThreadPool 

Just I show you simple example

def test_multithread_stringio_read_csv(self): # see gh-11786 max_row_range = 10000 num_files = 100 bytes_to_df = [ '\n'.join( ['%d,%d,%d' % (i, i, i) for i in range(max_row_range)] ).encode() for j in range(num_files)] files = [BytesIO(b) for b in bytes_to_df] # read all files in many threads pool = ThreadPool(8) results = pool.map(self.read_csv, files) first_result = results[0] for result in results: tm.assert_frame_equal(first_result, result) 

Here's the result I finally ended up using. It's a modified version of the classes by dgorissen above.

File: threadpool.py

from queue import Queue, Empty import threading from threading import Thread class Worker(Thread): _TIMEOUT = 2 """ Thread executing tasks from a given tasks queue. Thread is signalable, to exit """ def __init__(self, tasks, th_num): Thread.__init__(self) self.tasks = tasks self.daemon, self.th_num = True, th_num self.done = threading.Event() self.start() def run(self): while not self.done.is_set(): try: func, args, kwargs = self.tasks.get(block=True, timeout=self._TIMEOUT) try: func(*args, **kwargs) except Exception as e: print(e) finally: self.tasks.task_done() except Empty as e: pass return def signal_exit(self): """ Signal to thread to exit """ self.done.set() class ThreadPool: """Pool of threads consuming tasks from a queue""" def __init__(self, num_threads, tasks=[]): self.tasks = Queue(num_threads) self.workers = [] self.done = False self._init_workers(num_threads) for task in tasks: self.tasks.put(task) def _init_workers(self, num_threads): for i in range(num_threads): self.workers.append(Worker(self.tasks, i)) def add_task(self, func, *args, **kwargs): """Add a task to the queue""" self.tasks.put((func, args, kwargs)) def _close_all_threads(self): """ Signal all threads to exit and lose the references to them """ for workr in self.workers: workr.signal_exit() self.workers = [] def wait_completion(self): """Wait for completion of all the tasks in the queue""" self.tasks.join() def __del__(self): self._close_all_threads() def create_task(func, *args, **kwargs): return (func, args, kwargs) 

To use the pool

from random import randrange from time import sleep delays = [randrange(1, 10) for i in range(30)] def wait_delay(d): print('sleeping for (%d)sec' % d) sleep(d) pool = ThreadPool(20) for i, d in enumerate(delays): pool.add_task(wait_delay, d) pool.wait_completion() 

another way can be adding the process to thethread queue pool

import concurrent.futures with concurrent.futures.ThreadPoolExecutor(max_workers=cpus) as executor: for i in range(10): a = executor.submit(arg1, arg2,....) 

The overhead of creating the new processes is minimal, especially when it's just 4 of them. I doubt this is a performance hot spot of your application. Keep it simple, optimize where you have to and where profiling results point to.

There is no built in thread based pool. However, it can be very quick to implement a producer/consumer queue with the Queue class.

From: https://docs.python.org/2/library/queue.html

from threading import Thread from Queue import Queue def worker(): while True: item = q.get() do_work(item) q.task_done() q = Queue() for i in range(num_worker_threads): t = Thread(target=worker) t.daemon = True t.start() for item in source(): q.put(item) q.join() # block until all tasks are done 

If you don't mind executing other's code, here's mine:

Note: There is lot of extra code you may want to remove [added for better clarificaiton and demonstration how it works]

Note: Python naming conventions were used for method names and variable names instead of camelCase.

Working procedure:

1. MultiThread class will initiate with no of instances of threads by sharing lock, work queue, exit flag and results.
2. SingleThread will be started by MultiThread once it creates all instances.
3. We can add works using MultiThread (It will take care of locking).
4. SingleThreads will process work queue using a lock in middle.
5. Once your work is done, you can destroy all threads with shared boolean value.
6. Here, work can be anything. It can automatically import (uncomment import line) and process module using given arguments.
7. Results will be added to results and we can get using get_results

Code:

import threading import queue class SingleThread(threading.Thread): def __init__(self, name, work_queue, lock, exit_flag, results): threading.Thread.__init__(self) self.name = name self.work_queue = work_queue self.lock = lock self.exit_flag = exit_flag self.results = results def run(self): # print("Coming %s with parameters %s", self.name, self.exit_flag) while not self.exit_flag: # print(self.exit_flag) self.lock.acquire() if not self.work_queue.empty(): work = self.work_queue.get() module, operation, args, kwargs = work.module, work.operation, work.args, work.kwargs self.lock.release() print("Processing : " + operation + " with parameters " + str(args) + " and " + str(kwargs) + " by " + self.name + "\n") # module = __import__(module_name) result = str(getattr(module, operation)(*args, **kwargs)) print("Result : " + result + " for operation " + operation + " and input " + str(args) + " " + str(kwargs)) self.results.append(result) else: self.lock.release() # process_work_queue(self.work_queue) class MultiThread: def __init__(self, no_of_threads): self.exit_flag = bool_instance() self.queue_lock = threading.Lock() self.threads = [] self.work_queue = queue.Queue() self.results = [] for index in range(0, no_of_threads): thread = SingleThread("Thread" + str(index+1), self.work_queue, self.queue_lock, self.exit_flag, self.results) thread.start() self.threads.append(thread) def add_work(self, work): self.queue_lock.acquire() self.work_queue._put(work) self.queue_lock.release() def destroy(self): self.exit_flag.value = True for thread in self.threads: thread.join() def get_results(self): return self.results class Work: def __init__(self, module, operation, args, kwargs={}): self.module = module self.operation = operation self.args = args self.kwargs = kwargs class SimpleOperations: def sum(self, *args): return sum([int(arg) for arg in args]) @staticmethod def mul(a, b, c=0): return int(a) * int(b) + int(c) class bool_instance: def __init__(self, value=False): self.value = value def __setattr__(self, key, value): if key != "value": raise AttributeError("Only value can be set!") if not isinstance(value, bool): raise AttributeError("Only True/False can be set!") self.__dict__[key] = value # super.__setattr__(key, bool(value)) def __bool__(self): return self.value if __name__ == "__main__": multi_thread = MultiThread(5) multi_thread.add_work(Work(SimpleOperations(), "mul", [2, 3], {"c":4})) while True: data_input = input() if data_input == "": pass elif data_input == "break": break else: work = data_input.split() multi_thread.add_work(Work(SimpleOperations(), work[0], work[1:], {})) multi_thread.destroy() print(multi_thread.get_results())