Example 20-2 contains the implementation of flags.py, the first script we ran in Example 20-1. It’s not very interesting, but we’ll reuse most of its code and settings to implement the concurrent scripts, so it deserves some attention.

import time
from pathlib import Path
from typing import Callable

import httpx  

POP20_CC = ('CN IN US ID BR PK NG BD RU JP '
            'MX PH VN ET EG DE IR TR CD FR').split()  

BASE_URL = 'https://www.fluentpython.com/data/flags'  
DEST_DIR = Path('downloaded')                         

def save_flag(img: bytes, filename: str) -> None:     
    (DEST_DIR / filename).write_bytes(img)

def get_flag(cc: str) -> bytes:  
    url = f'{BASE_URL}/{cc}/{cc}.gif'.lower()
    resp = httpx.get(url, timeout=6.1,       
                     follow_redirects=True)  
    resp.raise_for_status()  
    return resp.content

def download_many(cc_list: list[str]) -> int:  
    for cc in sorted(cc_list):                 
        image = get_flag(cc)
        save_flag(image, f'{cc}.gif')
        print(cc, end=' ', flush=True)         
    return len(cc_list)

def main(downloader: Callable[[list[str]], int]) -> None:  
    DEST_DIR.mkdir(exist_ok=True)                          
    t0 = time.perf_counter()                               
    count = downloader(POP20_CC)
    elapsed = time.perf_counter() - t0
    print(f'\n{count} downloads in {elapsed:.2f}s')

if __name__ == '__main__':
    main(download_many)     

Import the httpx library. It’s not part of the standard library, so by convention the import goes after the standard library modules and a blank line.

List of the ISO 3166 country codes for the 20 most populous countries in order of decreasing population.

The directory with the flag images.⁴

Local directory where the images are saved.

Save the img bytes to filename in the DEST_DIR.

Given a country code, build the URL and download the image, returning the binary contents of the response.

It’s good practice to add a sensible timeout to network operations, to avoid blocking for several minutes for no good reason.

By default, HTTPX does not follow redirects.⁵

There’s no error handling in this script, but this method raises an exception if the HTTP status is not in the 2XX range—highly recommended to avoid silent failures.

download_many is the key function to compare with the concurrent implementations.

Loop over the list of country codes in alphabetical order, to make it easy to see that the ordering is preserved in the output; return the number of country codes downloaded.

Display one country code at a time in the same line so we can see progress as each download happens. The end=' ' argument replaces the usual line break at the end of each line printed with a space character, so all country codes are displayed progressively in the same line. The flush=True argument is needed because, by default, Python output is line buffered, meaning that Python only displays printed characters after a line break.

main must be called with the function that will make the downloads; that way, we can use main as a library function with other implementations of download_many in the threadpool and ascyncio examples.

Create DEST_DIR if needed; don’t raise an error if the directory exists.

Record and report the elapsed time after running the downloader function.

Call main with the download_many function.

There’s really nothing new to flags.py. It serves as a baseline for comparing the other scripts, and I used it as a library to avoid redundant code when implementing them. Now let’s see a reimplementation using concurrent.futures.

The main features of the concurrent.futures package are the ThreadPoolExecutor and ProcessPoolExecutor classes, which implement an API to submit callables for execution in different threads or processes, respectively. The classes transparently manage a pool of worker threads or processes, and queues to distribute jobs and collect results. But the interface is very high-level, and we don’t need to know about any of those details for a simple use case like our flag downloads.

Example 20-3 shows the easiest way to implement the downloads concurrently, using the ThreadPoolExecutor.map method.

from concurrent import futures

from flags import save_flag, get_flag, main  

def download_one(cc: str):  
    image = get_flag(cc)
    save_flag(image, f'{cc}.gif')
    print(cc, end=' ', flush=True)
    return cc

def download_many(cc_list: list[str]) -> int:
    with futures.ThreadPoolExecutor() as executor:         
        res = executor.map(download_one, sorted(cc_list))  

    return len(list(res))                                  

if __name__ == '__main__':
    main(download_many)  

Reuse some functions from the flags module (Example 20-2).

Function to download a single image; this is what each worker will execute.

Instantiate the ThreadPoolExecutor as a context manager; the executor​.__exit__ method will call executor.shutdown(wait=True), which will block until all threads are done.

The map method is similar to the map built-in, except that the download_one function will be called concurrently from multiple threads; it returns a generator that you can iterate to retrieve the value returned by each function call—in this case, each call to download_one will return a country code.

Return the number of results obtained. If any of the threaded calls raises an exception, that exception is raised here when the implicit next() call inside the list constructor tries to retrieve the corresponding return value from the iterator returned by executor.map.

Call the main function from the flags module, passing the concurrent version of download_many.

Note that the download_one function from Example 20-3 is essentially the body of the for loop in the download_many function from Example 20-2. This is a common refactoring when writing concurrent code: turning the body of a sequential for loop into a function to be called concurrently.

The ThreadPoolExecutor constructor takes several arguments not shown, but the first and most important one is max_workers, setting the maximum number of worker threads to be executed. When max_workers is None (the default), ThreadPool​Executor decides its value using the following expression—since Python 3.8:

max_workers = min(32, os.cpu_count() + 4)

The rationale is explained in the ThreadPoolExecutor documentation:

This default value preserves at least 5 workers for I/O bound tasks. It utilizes at most 32 CPU cores for CPU bound tasks which release the GIL. And it avoids using very large resources implicitly on many-core machines.

ThreadPoolExecutor now reuses idle worker threads before starting max_workers worker threads too.

To conclude: the computed default for max_workers is sensible, and ThreadPoolExecutor avoids starting new workers unnecessarily. Understanding the logic behind max_workers may help you decide when and how to set it yourself.

The library is called concurrency.futures, yet there are no futures to be seen in Example 20-3, so you may be wondering where they are. The next section explains.

Futures are core components of concurrent.futures and of asyncio, but as users of these libraries we sometimes don’t see them. Example 20-3 depends on futures behind the scenes, but the code I wrote does not touch them directly. This section is an overview of futures, with an example that shows them in action.

Since Python 3.4, there are two classes named Future in the standard library: concurrent.futures.Future and asyncio.Future. They serve the same purpose: an instance of either Future class represents a deferred computation that may or may not have completed. This is somewhat similar to the Deferred class in Twisted, the Future class in Tornado, and Promise in modern JavaScript.

Futures encapsulate pending operations so that we can put them in queues, check whether they are done, and retrieve results (or exceptions) when they become available.

An important thing to know about futures is that you and I should not create them: they are meant to be instantiated exclusively by the concurrency framework, be it concurrent.futures or asyncio. Here is why: a Future represents something that will eventually run, therefore it must be scheduled to run, and that’s the job of the framework. In particular, concurrent.futures.Future instances are created only as the result of submitting a callable for execution with a concurrent.futures.Executor subclass. For example, the Executor.submit() method takes a callable, schedules it to run, and returns a Future.

Application code is not supposed to change the state of a future: the concurrency framework changes the state of a future when the computation it represents is done, and we can’t control when that happens.

Both types of Future have a .done() method that is nonblocking and returns a Boolean that tells you whether the callable wrapped by that future has executed or not. However, instead of repeatedly asking whether a future is done, client code usually asks to be notified. That’s why both Future classes have an .add_done_callback() method: you give it a callable, and the callable will be invoked with the future as the single argument when the future is done. Be aware that the callback callable will run in the same worker thread or process that ran the function wrapped in the future.

There is also a .result() method, which works the same in both classes when the future is done: it returns the result of the callable, or re-raises whatever exception might have been thrown when the callable was executed. However, when the future is not done, the behavior of the result method is very different between the two flavors of Future. In a concurrency.futures.Future instance, invoking f.result() will block the caller’s thread until the result is ready. An optional timeout argument can be passed, and if the future is not done in the specified time, the result method raises TimeoutError. The asyncio.Future.result method does not support timeout, and await is the preferred way to get the result of futures in asyncio—but await doesn’t work with concurrency.futures.Future instances.

Several functions in both libraries return futures; others use them in their implementation in a way that is transparent to the user. An example of the latter is the Executor.map we saw in Example 20-3: it returns an iterator in which __next__ calls the result method of each future, so we get the results of the futures, and not the futures themselves.

To get a practical look at futures, we can rewrite Example 20-3 to use the concurrent.futures.as_completed function, which takes an iterable of futures and returns an iterator that yields futures as they are done.

Using futures.as_completed requires changes to the download_many function only. The higher-level executor.map call is replaced by two for loops: one to create and schedule the futures, the other to retrieve their results. While we are at it, we’ll add a few print calls to display each future before and after it’s done. Example 20-4 shows the code for a new download_many function. The code for download_many grew from 5 to 17 lines, but now we get to inspect the mysterious futures. The remaining functions are the same as in Example 20-3.

def download_many(cc_list: list[str]) -> int:
    cc_list = cc_list[:5]  
    with futures.ThreadPoolExecutor(max_workers=3) as executor:  
        to_do: list[futures.Future] = []
        for cc in sorted(cc_list):  
            future = executor.submit(download_one, cc)  
            to_do.append(future)  
            print(f'Scheduled for {cc}: {future}')  

        for count, future in enumerate(futures.as_completed(to_do), 1):  
            res: str = future.result()  
            print(f'{future} result: {res!r}')  

    return count

For this demonstration, use only the top five most populous countries.

Set max_workers to 3 so we can see pending futures in the output.

Iterate over country codes alphabetically, to make it clear that results will arrive out of order.

executor.submit schedules the callable to be executed, and returns a future representing this pending operation.

Store each future so we can later retrieve them with as_completed.

Display a message with the country code and the respective future.

as_completed yields futures as they are completed.

Get the result of this future.

Display the future and its result.

Note that the future.result() call will never block in this example because the future is coming out of as_completed. Example 20-5 shows the output of one run of Example 20-4.

$ python3 flags_threadpool_futures.py
Scheduled for BR: <Future at 0x100791518 state=running>  
Scheduled for CN: <Future at 0x100791710 state=running>
Scheduled for ID: <Future at 0x100791a90 state=running>
Scheduled for IN: <Future at 0x101807080 state=pending>  
Scheduled for US: <Future at 0x101807128 state=pending>
CN <Future at 0x100791710 state=finished returned str> result: 'CN'  
BR ID <Future at 0x100791518 state=finished returned str> result: 'BR'  
<Future at 0x100791a90 state=finished returned str> result: 'ID'
IN <Future at 0x101807080 state=finished returned str> result: 'IN'
US <Future at 0x101807128 state=finished returned str> result: 'US'

5 downloads in 0.70s

The futures are scheduled in alphabetical order; the repr() of a future shows its state: the first three are running, because there are three worker threads.

The last two futures are pending, waiting for worker threads.

The first CN here is the output of download_one in a worker thread; the rest of the line is the output of download_many.

Here, two threads output codes before download_many in the main thread can display the result of the first thread.

We saw two variants of the download script using concurrent.futures: one in Example 20-3 with ThreadPoolExecutor.map and one in Example 20-4 with futures.as_completed. If you are curious about the code for flags_asyncio.py, you may peek at Example 21-3 in Chapter 21, where it is explained.

Now let’s take a brief look at a simple way to work around the GIL for CPU-bound jobs using concurrent.futures.

Multicore Prime Checker Redux

In “Code for the Multicore Prime Checker” we studied procs.py, a script that checked the primality of some large numbers using multiprocessing. In Example 20-6 we solve the same problem in the proc_pool.py program using a ProcessPool​Executor. From the first import to the main() call at the end, procs.py has 43 nonblank lines of code, and proc_pool.py has 31—28% shorter.

import sys
from concurrent import futures  
from time import perf_counter
from typing import NamedTuple

from primes import is_prime, NUMBERS

class PrimeResult(NamedTuple):  
    n: int
    flag: bool
    elapsed: float

def check(n: int) -> PrimeResult:
    t0 = perf_counter()
    res = is_prime(n)
    return PrimeResult(n, res, perf_counter() - t0)

def main() -> None:
    if len(sys.argv) < 2:
        workers = None      
    else:
        workers = int(sys.argv[1])

    executor = futures.ProcessPoolExecutor(workers)  
    actual_workers = executor._max_workers  # type: ignore  

    print(f'Checking {len(NUMBERS)} numbers with {actual_workers} processes:')

    t0 = perf_counter()

    numbers = sorted(NUMBERS, reverse=True)  
    with executor:  
        for n, prime, elapsed in executor.map(check, numbers):  
            label = 'P' if prime else ' '
            print(f'{n:16}  {label} {elapsed:9.6f}s')

    time = perf_counter() - t0
    print(f'Total time: {time:.2f}s')

if __name__ == '__main__':
    main()

No need to import multiprocessing, SimpleQueue etc.; concurrent.futures hides all that.

The PrimeResult tuple and the check function are the same as we saw in procs.py, but we don’t need the queues and the worker function anymore.

Instead of deciding ourselves how many workers to use if no command-line argument was given, we set workers to None and let the ProcessPoolExecutor decide.

Here I build the ProcessPoolExecutor before the with block in ➐ so that I can display the actual number of workers in the next line.

_max_workers is an undocumented instance attribute of a ProcessPoolExecutor. I decided to use it to show the number of workers when the workers variable is None. Mypy correctly complains when I access it, so I put the type: ignore comment to silence it.

Sort the numbers to be checked in descending order. This will expose a difference in the behavior of proc_pool.py when compared with procs.py. See the explanation after this example.

Use the executor as a context manager.

The executor.map call returns the PrimeResult instances returned by check in the same order as the numbers arguments.

If you run Example 20-6, you’ll see the results appearing in strict descending order, as shown in Example 20-7. In contrast, the ordering of the output of procs.py (shown in “Process-Based Solution”) is heavily influenced by the difficulty in checking whether each number is a prime. For example, procs.py shows the result for 7777777777777777 near the top, because it has a low divisor, 7, so is_prime quickly determines it’s not a prime.

In contrast, 7777777536340681 is 88191709², so is_prime will take much longer to determine that it’s a composite number, and even longer to find out that 7777777777777753 is prime—therefore both of these numbers appear near the end of the output of procs.py.

Running proc_pool.py, you’ll observe not only the descending order of the results, but also that the program will appear to be stuck after showing the result for 9999999999999999.

$ ./proc_pool.py
Checking 20 numbers with 12 processes:
9999999999999999     0.000024s  
9999999999999917  P  9.500677s  
7777777777777777     0.000022s  
7777777777777753  P  8.976933s
7777777536340681     8.896149s
6666667141414921     8.537621s
6666666666666719  P  8.548641s
6666666666666666     0.000002s
5555555555555555     0.000017s
5555555555555503  P  8.214086s
5555553133149889     8.067247s
4444444488888889     7.546234s
4444444444444444     0.000002s
4444444444444423  P  7.622370s
3333335652092209     6.724649s
3333333333333333     0.000018s
3333333333333301  P  6.655039s
 299593572317531  P  2.072723s
 142702110479723  P  1.461840s
               2  P  0.000001s
Total time: 9.65s

This line appears very quickly.

This line takes more than 9.5s to show up.

All the remaining lines appear almost immediately.

Here is why proc_pool.py behaves in that way:

• As mentioned before, executor.map(check, numbers) returns the result in the same order as the numbers are given.

• By default, proc_pool.py uses as many workers as there are CPUs—it’s what ProcessPoolExecutor does when max_workers is None. That’s 12 processes in this laptop.

• Because we are submitting numbers in descending order, the first is 9999999999999999; with 9 as a divisor, it returns quickly.

• The second number is 9999999999999917, the largest prime in the sample. This will take longer than all the others to check.

• Meanwhile, the remaining 11 processes will be checking other numbers, which are either primes or composites with large factors, or composites with very small factors.

• When the worker in charge of 9999999999999917 finally determines that’s a prime, all the other processes have completed their last jobs, so the results appear immediately after.

Understanding how concurrent programs behave is not straightforward, so here’s is a second experiment that may help you visualize the operation of Executor.map.

Error Handling in the flags2 Examples

The common strategy in all three examples to deal with HTTP errors is that 404 errors (not found) are handled by the function in charge of downloading a single file (download_one). Any other exception propagates to be handled by the download_many function or the supervisor coroutine—in the asyncio example.

Once more, we’ll start by studying the sequential code, which is easier to follow—and mostly reused by the thread pool script. Example 20-14 shows the functions that perform the actual downloads in the flags2_sequential.py and flags2_threadpool.py scripts.

from collections import Counter
from http import HTTPStatus

import httpx
import tqdm  # type: ignore  

from flags2_common import main, save_flag, DownloadStatus  

DEFAULT_CONCUR_REQ = 1
MAX_CONCUR_REQ = 1

def get_flag(base_url: str, cc: str) -> bytes:
    url = f'{base_url}/{cc}/{cc}.gif'.lower()
    resp = httpx.get(url, timeout=3.1, follow_redirects=True)
    resp.raise_for_status()  
    return resp.content

def download_one(cc: str, base_url: str, verbose: bool = False) -> DownloadStatus:
    try:
        image = get_flag(base_url, cc)
    except httpx.HTTPStatusError as exc:  
        res = exc.response
        if res.status_code == HTTPStatus.NOT_FOUND:
            status = DownloadStatus.NOT_FOUND  
            msg = f'not found: {res.url}'
        else:
            raise  
    else:
        save_flag(image, f'{cc}.gif')
        status = DownloadStatus.OK
        msg = 'OK'

    if verbose:  
        print(cc, msg)

    return status

Import the tqdm progress-bar display library, and tell Mypy to skip checking it.⁷

Import a couple of functions and an Enum from the flags2_common module.

Raises HTTPStetusError if the HTTP status code is not in range(200, 300).

download_one catches HTTPStatusError to handle HTTP code 404 specifically…

…by setting its local status to DownloadStatus.NOT_FOUND; DownloadStatus is an Enum imported from flags2_common.py.

Any other HTTPStatusError exception is re-raised to propagate to the caller.

If the -v/--verbose command-line option is set, the country code and status message are displayed; this is how you’ll see progress in verbose mode.

Example 20-15 lists the sequential version of the download_many function. This code is straightforward, but it’s worth studying to contrast with the concurrent versions coming up. Focus on how it reports progress, handles errors, and tallies downloads.

def download_many(cc_list: list[str],
                  base_url: str,
                  verbose: bool,
                  _unused_concur_req: int) -> Counter[DownloadStatus]:
    counter: Counter[DownloadStatus] = Counter()  
    cc_iter = sorted(cc_list)  
    if not verbose:
        cc_iter = tqdm.tqdm(cc_iter)  
    for cc in cc_iter:
        try:
            status = download_one(cc, base_url, verbose)  
        except httpx.HTTPStatusError as exc:  
            error_msg = 'HTTP error {resp.status_code} - {resp.reason_phrase}'
            error_msg = error_msg.format(resp=exc.response)
        except httpx.RequestError as exc:  
            error_msg = f'{exc} {type(exc)}'.strip()
        except KeyboardInterrupt:  
            break
        else:  
            error_msg = ''

        if error_msg:
            status = DownloadStatus.ERROR  
        counter[status] += 1           
        if verbose and error_msg:      
            print(f'{cc} error: {error_msg}')

    return counter  

This Counter will tally the different download outcomes: DownloadStatus.OK, DownloadStatus.NOT_FOUND, or DownloadStatus.ERROR.

cc_iter holds the list of the country codes received as arguments, ordered alphabetically.

If not running in verbose mode, cc_iter is passed to tqdm, which returns an iterator yielding the items in cc_iter while also animating the progress bar.

Make successive calls to download_one.

HTTP status code exceptions raised by get_flag and not handled by download_one are handled here.

Other network-related exceptions are handled here. Any other exception will abort the script, because the flags2_common.main function that calls download_many has no try/except.

Exit the loop if the user hits Ctrl-C.

If no exception escaped download_one, clear the error message.

If there was an error, set the local status accordingly.

Increment the counter for that status.

In verbose mode, display the error message for the current country code, if any.

Return counter so that main can display the numbers in the final report.

We’ll now study the refactored thread pool example, flags2_threadpool.py.

Using futures.as_completed

In order to integrate the tqdm progress bar and handle errors on each request, the flags2_threadpool.py script uses futures.ThreadPoolExecutor with the futures.as_completed function we’ve already seen. Example 20-16 is the full listing of flags2_threadpool.py. Only the download_many function is implemented; the other functions are reused from flags2_common.py and flags2_sequential.py.

from collections import Counter
from concurrent.futures import ThreadPoolExecutor, as_completed

import httpx
import tqdm  # type: ignore

from flags2_common import main, DownloadStatus
from flags2_sequential import download_one  

DEFAULT_CONCUR_REQ = 30  
MAX_CONCUR_REQ = 1000  


def download_many(cc_list: list[str],
                  base_url: str,
                  verbose: bool,
                  concur_req: int) -> Counter[DownloadStatus]:
    counter: Counter[DownloadStatus] = Counter()
    with ThreadPoolExecutor(max_workers=concur_req) as executor:  
        to_do_map = {}  
        for cc in sorted(cc_list):  
            future = executor.submit(download_one, cc,
                                     base_url, verbose)  
            to_do_map[future] = cc  
        done_iter = as_completed(to_do_map)  
        if not verbose:
            done_iter = tqdm.tqdm(done_iter, total=len(cc_list))  
        for future in done_iter:  
            try:
                status = future.result()  
            except httpx.HTTPStatusError as exc:  
                error_msg = 'HTTP error {resp.status_code} - {resp.reason_phrase}'
                error_msg = error_msg.format(resp=exc.response)
            except httpx.RequestError as exc:
                error_msg = f'{exc} {type(exc)}'.strip()
            except KeyboardInterrupt:
                break
            else:
                error_msg = ''

            if error_msg:
                status = DownloadStatus.ERROR
            counter[status] += 1
            if verbose and error_msg:
                cc = to_do_map[future]  
                print(f'{cc} error: {error_msg}')

    return counter


if __name__ == '__main__':
    main(download_many, DEFAULT_CONCUR_REQ, MAX_CONCUR_REQ)

Reuse download_one from flags2_sequential (Example 20-14).

If the -m/--max_req command-line option is not given, this will be the maximum number of concurrent requests, implemented as the size of the thread pool; the actual number may be smaller if the number of flags to download is smaller.

MAX_CONCUR_REQ caps the maximum number of concurrent requests regardless of the number of flags to download or the -m/--max_req command-line option. It’s a safety precaution to avoid launching too many threads with their significant memory overhead.

Create the executor with max_workers set to concur_req, computed by the main function as the smaller of: MAX_CONCUR_REQ, the length of cc_list, or the value of the -m/--max_req command-line option. This avoids creating more threads than necessary.

This dict will map each Future instance—representing one download—with the respective country code for error reporting.

Iterate over the list of country codes in alphabetical order. The order of the results will depend on the timing of the HTTP responses more than anything, but if the size of the thread pool (given by concur_req) is much smaller than len(cc_list), you may notice the downloads batched alphabetically.

Each call to executor.submit schedules the execution of one callable and returns a Future instance. The first argument is the callable, the rest are the arguments it will receive.

Store the future and the country code in the dict.

futures.as_completed returns an iterator that yields futures as each task is done.

If not in verbose mode, wrap the result of as_completed with the tqdm function to display the progress bar; because done_iter has no len, we must tell tqdm what is the expected number of items as the total= argument, so tqdm can estimate the work remaining.

Iterate over the futures as they are completed.

Calling the result method on a future either returns the value returned by the callable, or raises whatever exception was caught when the callable was executed. This method may block waiting for a resolution, but not in this example because as_completed only returns futures that are done.

Handle the potential exceptions; the rest of this function is identical to the sequential download_many in Example 20-15), except for the next callout.

To provide context for the error message, retrieve the country code from the to_do_map using the current future as key. This was not necessary in the sequential version because we were iterating over the list of country codes, so we knew the current cc; here we are iterating over the futures.

Python threads are well suited for I/O-intensive applications, and the concurrent.futures package makes it relatively simple to use for certain use cases. With ProcessPoolExecutor, you can also solve CPU-intensive problems on multiple cores—if the computations are “embarrassingly parallel”. This concludes our basic introduction to concurrent.futures.