Python ThreadPoolExecutor
Article No - 37
Python ThreadPoolExecutor
Overview:
- Thread: A thread in computer science is short for a thread of execution. Threads are a way for a program to divide (termed "split") itself into two or more simultaneously (or pseudo-simultaneously) running tasks. Threads are lightweight, in terms of the system resource they consume, as compared with processes.
- Suppose you have to create a large number of threads for our multithreaded task.
- There can be many performances issue due to too many threads. A major issue is a throughput limit.
- We can solve this problem by creating a pool of threads.
- A thread pool may be defined as the group of pre-instantiated and idle threads, which stand ready to be given work.
- Creating a thread pool is preferred over instantiating new threads for every task when we need to do a large number of tasks.
Python module: Concurrent.futures
- The concurrent.futures module was added in Python 3.2.
- Python standard library includes the concurrent.futures module.
- According to the Python documentation, it provides the developer with a high-level interface for asynchronously executing callables. Basically concurrent.futures is an abstraction layer on top of Python threading and multiprocessing modules that simplifies using them.
- It has 2 main classes which provide concurrent execution:
1. ThreadPoolExecutor
2. ProcessPoolExecutor
- Concurrent.futures includes an abstract class called Executor. ThreadPoolExecutor and ProcessPoolExecutor are two subclasses of Executor. Both the subclass implements the same interface which is defined by the abstract class Executor.
import concurrent.futures
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import threading
import time
ThreadPoolExecutor:
ThreadPoolExecutor = Thread + Pool + Executor.
Pool:
The Pool portion is where it starts to get interesting. This object is going to create a pool of threads, each of which can run concurrently.
Executor :
ThreadPoolExecutoris anExecutorsubclass that uses a pool of threads to execute calls asynchronously.- With the help of concurrent.futures module and its concrete subclass Executor, we can easily create a pool of threads.
ThreadPoolExecutor(max_workers=None,thread_name_prefix='',initializer=None,initargs=())Parameters:
- max_workers: number of threads we want in the pool. If max_workers is
Noneor not given, it will default to the number of processors on the machine, multiplied by 5. (Change in python 3.5)
- thread_name_prefix: Optional,allow users to control the
threading.Threadnames for worker threads created by the pool for easier debugging.
- initializer: optional callable that is called at the start of each worker thread.
- initargs: a tuple of arguments passed to the initializer.
Example: 1
Only max_worker passed as a parameter.
import concurrent.futures
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import threading
import time
def twiceOfNumber(x):
curr_thread = threading.current_thread()
time.sleep(x)
return curr_thread.name #return threadPoolExecuter object
if __name__ == '__main__':
executer = ThreadPoolExecutor(max_workers=5)
resultList = []
for i in range(1,11):
resultList.append(executer.submit(twiceOfNumber,i))
print([future.result() for future in resultList])
['ThreadPoolExecutor-4_0', 'ThreadPoolExecutor-4_1', 'ThreadPoolExecutor-4_2', 'ThreadPoolExecutor-4_3', 'ThreadPoolExecutor-4_4', 'ThreadPoolExecutor-4_0', 'ThreadPoolExecutor-4_1', 'ThreadPoolExecutor-4_2', 'ThreadPoolExecutor-4_3', 'ThreadPoolExecutor-4_4']
Example: 2
max_worker and thread_name_prefix passed as a parameter.
import concurrent.futures
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import threading
import time
def twiceOfNumber(x):
curr_thread = threading.current_thread()
time.sleep(x)
return curr_thread.name #return threadPoolExecuter object but name replaced with thread_name_prefix
if __name__ == '__main__':
executer = ThreadPoolExecutor(max_workers=5,thread_name_prefix='threads')
resultList = []
for i in range(1,11):
resultList.append(executer.submit(twiceOfNumber,i))
print([future.result() for future in resultList])
Output
['threads_0', 'threads_1', 'threads_2', 'threads_3', 'threads_4', 'threads_0', 'threads_1', 'threads_2', 'threads_3', 'threads_4']
Example: 3
ThreadPoolExecutor have arguments named initializer and initargs which can be used to execute some callable before each of Thread/Process execution.import concurrent.futures
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import threading
import time
def initializer(x):
print('Initilizing task enviroment : %d'%x)
def twiceOfNumber(x):
curr_thread = threading.current_thread() # check the current thread
time.sleep(x)
return curr_thread.name # check the current thread name
if __name__ == '__main__':
executer=ThreadPoolExecutor(max_workers=5, thread_name_prefix = 'Thread', initializer=initializer, initargs=(10,))
result = executer.map(twiceOfNumber, range(1,11))
print(list(result))
Output
Initilizing task enviroment : 10
Initilizing task enviroment : 10
Initilizing task enviroment : 10
Initilizing task enviroment : 10
Initilizing task enviroment : 10
['Thread_0', 'Thread_1', 'Thread_2', 'Thread_3', 'Thread_4', 'Thread_0', 'Thread_1', 'Thread_2', 'Thread_3', 'Thread_4']
Example: 4
import concurrent.futures
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import threading
import time
def twiceOfNumber(x):
curr_thread = threading.current_thread()
time.sleep(x)
return x*2 #return twice of number
if __name__ == '__main__':
executer=ThreadPoolExecutor(max_workers=5, thread_name_prefix = 'Thread')
futures = []
for i in range(1,11):
futures.append(executer.submit(twiceOfNumber, i)) # return a future obj that is stored in futures list
print([future.result() for future in futures]) # accessing futures list getting the value using result method
Output
[2, 4, 6, 8, 10, 12, 14, 16, 18, 20]
Two way to submit the task to the executor pool:
- submit(fn,*args,**kwargs)
- map(func, *iterable, timeout=none, chunksize=1)
submit () method:
Return a future object and this future object has a method called result() which returns the calculated result. When a task finishes, the ThreadPoolExecutor sets the value to the future object.
(used in the example1, example2,example4).
map() method:
Similar to
map(func, *iterables) except, func is executed asynchronously and several calls to func may be made concurrently.map() method has argument named timeout which except time in seconds and if provided then iterator will raise concurrent.futures.TimeoutError if after that many second task has still not completed. (used in the Example3 and example5)
shutdown(wait=true) :
- ThreadPoolExecutor class method
- This function sends a signal to the executor that it should free any resources that it is using when the currently pending futures(obj) are done executing.
- if executor.submit() or executor.map() used after the shutdown() then shutdown will raise RunTimeError.
- one parameter: wait, which take the boolean value (true/false)
- wait=true, this method will not return until all the pending futures are done executing and the resources associated with the executor have been freed.
- wait=false, this method will return immediately and the resources associated with the executor will be freed when all pending futures are done executing.
- we can use ThreadPoolExecutor with 'with' keyword (context manager) that automatically calls the shutdown() method.
Example: 5
ThreadPoolExecutor with the context manager
import concurrent.futures
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import threading
import time
def concat(x):
curr_thread = threading.current_thread()
return "hello"+str(x) #string concatenation
if __name__ == '__main__':
with ThreadPoolExecutor(max_workers=5, thread_name_prefix = 'Thread') as executer:
result=executer.map(concat,range(1,10))
print(list(result))
Output
['hello1', 'hello2', 'hello3', 'hello4', 'hello5', 'hello6', 'hello7', 'hello8', 'hello9']
Comments