Concurrency in Python - Pool of Threads

Suppose we had to create a large number of threads for our multithreaded tasks. It would be computationally most expensive as there can be many performance issues, due to too many threads. A major issue could be in the throughput getting pmited. 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 thread pool is preferred over instantiating new threads for every task when we need to do large number of tasks. A thread pool can manage concurrent execution of large number of threads as follows −

If a thread in a thread pool completes its execution then that thread can be reused.

If a thread is terminated, another thread will be created to replace that thread.

Python Module – Concurrent.futures

Python standard pbrary includes the concurrent.futures module. This module was added in Python 3.2 for providing the developers a high-level interface for launching asynchronous tasks. It is an abstraction layer on the top of Python’s threading and multiprocessing modules for providing the interface for running the tasks using pool of thread or processes.

In our subsequent sections, we will learn about the different classes of the concurrent.futures module.

Executor Class

Executoris an abstract class of the concurrent.futures Python module. It cannot be used directly and we need to use one of the following concrete subclasses −

ThreadPoolExecutor

ProcessPoolExecutor

ThreadPoolExecutor – A Concrete Subclass

It is one of the concrete subclasses of the Executor class. The subclass uses multi-threading and we get a pool of thread for submitting the tasks. This pool assigns tasks to the available threads and schedules them to run.

How to create a ThreadPoolExecutor?

With the help of concurrent.futures module and its concrete subclass Executor, we can easily create a pool of threads. For this, we need to construct a ThreadPoolExecutor with the number of threads we want in the pool. By default, the number is 5. Then we can submit a task to the thread pool. When we submit() a task, we get back a Future. The Future object has a method called done(), which tells if the future has resolved. With this, a value has been set for that particular future object. When a task finishes, the thread pool executor sets the value to the future object.

Example

from concurrent.futures import ThreadPoolExecutor
from time import sleep
def task(message):
   sleep(2)
   return message

def main():
   executor = ThreadPoolExecutor(5)
   future = executor.submit(task, ("Completed"))
   print(future.done())
   sleep(2)
   print(future.done())
   print(future.result())
if __name__ ==  __main__ :
main()

Output

False
True
Completed

In the above example, a ThreadPoolExecutor has been constructed with 5 threads. Then a task, which will wait for 2 seconds before giving the message, is submitted to the thread pool executor. As seen from the output, the task does not complete until 2 seconds, so the first call to done() will return False. After 2 seconds, the task is done and we get the result of the future by calpng the result() method on it.

Instantiating ThreadPoolExecutor – Context Manager

Another way to instantiate ThreadPoolExecutor is with the help of context manager. It works similar to the method used in the above example. The main advantage of using context manager is that it looks syntactically good. The instantiation can be done with the help of the following code −

with ThreadPoolExecutor(max_workers = 5) as executor

Example

The following example is borrowed from the Python docs. In this example, first of all the concurrent.futures module has to be imported. Then a function named load_url() is created which will load the requested url. The function then creates ThreadPoolExecutor with the 5 threads in the pool. The ThreadPoolExecutor has been utipzed as context manager. We can get the result of the future by calpng the result() method on it.

import concurrent.futures
import urlpb.request

URLS = [ http://www.foxnews.com/ ,
    http://www.cnn.com/ ,
    http://europe.wsj.com/ ,
    http://www.bbc.co.uk/ ,
    http://some-made-up-domain.com/ ]

def load_url(url, timeout):
   with urlpb.request.urlopen(url, timeout = timeout) as conn:
   return conn.read()

with concurrent.futures.ThreadPoolExecutor(max_workers = 5) as executor:

   future_to_url = {executor.submit(load_url, url, 60): url for url in URLS}
   for future in concurrent.futures.as_completed(future_to_url):
   url = future_to_url[future]
   try:
      data = future.result()
   except Exception as exc:
      print( %r generated an exception: %s  % (url, exc))
   else:
      print( %r page is %d bytes  % (url, len(data)))

Output

Following would be the output of the above Python script −

 http://some-made-up-domain.com/  generated an exception: <urlopen error [Errno 11004] getaddrinfo failed>
 http://www.foxnews.com/  page is 229313 bytes
 http://www.cnn.com/  page is 168933 bytes
 http://www.bbc.co.uk/  page is 283893 bytes
 http://europe.wsj.com/  page is 938109 bytes

Use of Executor.map() function

The Python map() function is widely used in a number of tasks. One such task is to apply a certain function to every element within iterables. Similarly, we can map all the elements of an iterator to a function and submit these as independent jobs to out ThreadPoolExecutor. Consider the following example of Python script to understand how the function works.

Example

In this example below, the map function is used to apply the square() function to every value in the values array.

from concurrent.futures import ThreadPoolExecutor
from concurrent.futures import as_completed
values = [2,3,4,5]
def square(n):
   return n * n
def main():
   with ThreadPoolExecutor(max_workers = 3) as executor:
      results = executor.map(square, values)
for result in results:
      print(result)
if __name__ ==  __main__ :
   main()

Output

The above Python script generates the following output −

4
9
16
25