Skip to content
wannacode

A pool of threads and tasks

In short: A thread pool is a few workers in front of a task queue instead of a new thread per task; in Python this is ThreadPoolExecutor, and ProcessPoolExecutor for computation.

Problem

A thread for every image

The shop starts a sale. Sellers upload images of their products, and every image needs a thumbnail for the product list. Making a thumbnail is work that computes: it goes through every pixel (see work that waits and work that computes).

The first version starts a thread for every uploaded image:

threads = []
for image in uploads:
    thread = threading.Thread(target=save_thumbnail, args=(image,))
    thread.start()
    threads.append(thread)

With 20 images everything works on your machine: "Done: 20 thumbnails". In the first hour of the sale 10,000 come at once.

  • Every thread needs its own stack and a place in the operating system. Thousands of threads eat the memory, and near the system's limit start() fails with RuntimeError: can't start new thread.
  • Even if there is memory, because of the GIL only one thread computes at a time. The other 9,999 wait and take turns.

The server does not slow down; it falls over. And a server that is down does not serve the buyers either.

The idea

A few workers and a queue

A thread pool is a few workers, created once, and a queue of tasks in front of them. A task is "shrink this image". A free worker takes the next task from the queue, finishes it and comes back for a new one.

  • The number of workers is constant. 10,000 images do not make 10,000 threads, but 10,000 tasks in the queue. A task is a small object; a thread is expensive.
  • The workers are reused. A thread is not created and destroyed for every image.
  • The queue absorbs the peak. When uploads come faster than the workers can handle, the tasks wait instead of bringing the server down.

This is producer and consumer: the code that accepts uploads puts tasks into the queue, and the workers are the consumers.

You already met the pool in the lesson on threads: ThreadPoolExecutor, submit and Future.result(). Here we will see what happens inside, how to order the results, what happens to errors and how many workers you need.

See it

Three workers

Making a thumbnail here is a pause, 2 ms for every KB, and the size comes from the name of the image. That way the output is the same on every run:

# Making a thumbnail: here it is a pause, and the size comes from the name.
def thumbnail(image):
    if image.startswith("broken"):
        raise ValueError("corrupt file")
    kb = 20 + 3 * len(image)
    time.sleep(2 * kb / 1000)
    return kb

Eight images, a pool of three workers:

with ThreadPoolExecutor(max_workers=3) as pool:
    sizes = list(pool.map(thumbnail, images))
total = 0
for image, kb in zip(images, sizes):
    print(f"{image}: {kb} KB")
    total += kb
print(f"Total: {total} KB")
Step by step, in words
  1. 8 images wait. The pool is 3 threads: ThreadPoolExecutor(max_workers=3).
  2. map turns every image into a task in the queue. The three threads take the first three.
  3. Thread 3 finishes first: hat is the third image, and its result goes to the third place. The worker does not stop: it takes the next task.
  4. The queue is empty. Thread 1 is done and has nothing to take, so it waits. The others finish their last images.
  5. map returns the sizes in the order of the images, not in the order they finished. Total 379 KB; the end of with closes the pool.
images = ["boots.jpg", "jacket.jpg", "hat.jpg", "scarf.jpg", "backpack.jpg", "gloves.jpg", "watch.jpg", "cap.jpg"]
with ThreadPoolExecutor(max_workers=3) as pool:
    sizes = list(pool.map(thumbnail, images))
total = 0
for image, kb in zip(images, sizes):
    print(f"{image}: {kb} KB")
    total += kb
print(f"Total: {total} KB")
Output
boots.jpg: 47 KB
jacket.jpg: 50 KB
hat.jpg: 41 KB
scarf.jpg: 47 KB
backpack.jpg: 56 KB
gloves.jpg: 50 KB
watch.jpg: 47 KB
cap.jpg: 41 KB
Total: 379 KB
  • ThreadPoolExecutor(max_workers=3) makes three threads and a queue in front of them.
  • map makes a task for every image and returns the results in the order of the images, whatever finished first. If you do not wait for all of them, you take them one by one from map: it waits for exactly the next one.
  • with closes the pool at the end: it takes no new tasks, waits for the started ones and stops the threads.

Threads help here because making a thumbnail is a pause. A real thumbnail computes, and because of the GIL threads do not compute at the same time. For that there are processes, a bit further down.

More precisely, if you are curious

Without max_workers the pool has min(32, os.cpu_count() + 4) threads. Its queue is endless: if uploads always come faster, it grows until the memory runs out.

How it is in the other languages: in Java the pool is an ExecutorService (there threads really compute in parallel), in C# there is one pool for the whole program, and Node has no ready-made pool of Workers.

Try it

Find the bug

A colleague rewrote the thumbnails with a pool, and the sale did not get faster. Six images take as long as without a pool:

start = time.monotonic()
sizes = []
with ThreadPoolExecutor(max_workers=3) as pool:
    for image in images:
        size = pool.submit(thumbnail, image)
        sizes.append(size.result())
print(f"Took over 0.5 s: {time.monotonic() - start > 0.5}")
Output
Took over 0.5 s: True

Pick the line with the bug.

The idea

Results in input order

Tasks finish in an order you do not know: it depends on the images, on the load, on which worker was free. So never print "in the order they finished" if you want the same output every time.

Two ways give the same result on every run:

  • In input order. You keep the promises in a list in the order of the tasks and wait for them in that order. That is how the code above works.
  • A sum or a count. If you only need the total, the order does not matter: 47 + 50 is the same as 50 + 47.

Sometimes you want the results in the order they finished: for a "3 of 8 done" bar, or to show the first ready images. as_completed gives them that way. Then you only collect things that do not depend on the order:

count = 0
total = 0
with ThreadPoolExecutor(max_workers=3) as pool:
    futures = [pool.submit(thumbnail, image) for image in images]
    for future in as_completed(futures):
        total += future.result()
        count += 1
print(f"Done: {count} of {len(images)}, total: {total} KB")
Output
Done: 8 of 8, total: 379 KB
as_completed gives every Future as soon as it is done. result() on it does not wait.

The idea

An error inside a task

One of the images is corrupt, and thumbnail throws an exception. It does not come out in the thread that started the task: that thread has long moved on. The pool catches it and keeps it in the promise of the result. It comes out when you wait for that result:

images = ["boots.jpg", "jacket.jpg", "broken.jpg", "hat.jpg"]
with ThreadPoolExecutor(max_workers=3) as pool:
    futures = [pool.submit(thumbnail, image) for image in images]
for image, future in zip(images, futures):
    try:
        print(f"{image}: {future.result()} KB")
    except ValueError as error:
        print(f"{image}: error: {error}")
Output
boots.jpg: 47 KB
jacket.jpg: 50 KB
broken.jpg: error: corrupt file
hat.jpg: 41 KB
result() raises the same exception again (ValueError). The other images do not suffer. With pool.map the error comes out when the loop reaches the corrupt image, and the results after it are lost. So if you expect errors, use submit and result() for each one.

Try it

Where did the error go

The second image is corrupt. What does the program print?

images = ["boots.jpg", "broken.jpg", "hat.jpg"]
with ThreadPoolExecutor(max_workers=3) as pool:
    for image in images:
        pool.submit(thumbnail, image)
print(f"Done: {len(images)} images")
Answer

The idea

How many workers

It depends on what they do:

  • Work that computes (thumbnails, PDFs, compression): as many as the cores. A ninth worker on 8 cores has nowhere to run; it only takes turns with the others.
  • Work that waits (e-mail, payment, database): more. While one worker waits for the network, the core is free for another. If a task waits W ms and computes C ms, about 1 + W / C workers fit on a core.
def pool_size(cores, wait_ms, compute_ms):
    return cores * (1 + wait_ms // compute_ms)




print(f"Thumbnails (computing only): {pool_size(8, 0, 50)}")
print(f"E-mails (90 ms waiting, 10 ms computing): {pool_size(8, 90, 10)}")
Output
Thumbnails (computing only): 8
E-mails (90 ms waiting, 10 ms computing): 80

The number of cores comes from os.cpu_count(). The formula is a start, not a law: measure. If the e-mails go through one server with a limit of 20 connections, 80 workers will only wait in a queue in front of it.

Python has a catch: because of the GIL, threads do not compute at the same time. For work that computes the pool is made of processes: ProcessPoolExecutor, every process with its own Python and its own GIL. Tasks and results travel between the processes as copies:

from concurrent.futures import ProcessPoolExecutor




def checksum(invoice):
    return sum((invoice * 31 + i) % 256 for i in range(200_000))




if __name__ == "__main__":
    invoices = [1001, 1002, 1003, 1004]
    with ProcessPoolExecutor(max_workers=4) as pool:
        sums = list(pool.map(checksum, invoices))
    for invoice, total in zip(invoices, sums):
        print(f"Invoice {invoice}: {total}")
Output
Invoice 1001: 25497376
Invoice 1002: 25499360
Invoice 1003: 25501344
Invoice 1004: 25503328

if __name__ == "__main__": keeps the main code from running again in every process: on macOS and Windows a process starts by loading the file from the beginning.

Try it

Fill in

How many workers for a task that waits waitMs and computes computeMs?

def pool_size(cores, wait_ms, compute_ms):
    return cores * (1 + )
More precisely, if you are curious

For work that waits Python also has asyncio: one thread waits for thousands of requests (the lesson async). The thread pool stays for libraries that wait without being able to await.

Exercise

The invoices of the sale

Solve the task in your editor.

After the sale every order needs an invoice. invoice(amount) makes the PDF (here a pause of 50 ms) and returns the amount with 20% VAT. An order with an amount of 0 or less is invalid, and the function throws an exception with the text "invalid amount".

The orders are 1001 to 1008 with amounts €50, €125, €80, €0, €240, €35, €90 and €15. Make the invoices in a pool of 4 workers. Print them in the order of the orders, and the invalid one with the text of the error. At the end: the sum of the good invoices and how many failed.

Output
Order 1001: €60
Order 1002: €150
Order 1003: €96
Order 1004: error: invalid amount
Order 1005: €288
Order 1006: €42
Order 1007: €108
Order 1008: €18
Total: €762, failed: 1

More tasks

  • A bounded queue. Make the pool keep at most 10 tasks waiting in the queue. What happens to the eleventh? (threading.BoundedSemaphore around submit)
  • Progress. Print "3 of 8 done" after every finished invoice, and at the end the ordered list again.
  • A timeout. An invoice that takes over 1 second counts as failed, without stopping the others.

Test

Check yourself

1. Why a pool of a few threads, and not a thread for every uploaded image?
2. A pool of 3 workers shrinks 8 images. The third one finishes first. Where is its size when you read them in list order with get, result, Promise.all or sizes[i]?
3. A task in the pool throws an exception. When do you see it?
4. The server has 8 cores. Every e-mail waits 90 ms for the network and computes for 10 ms. About how many workers keep the cores busy?
5. How many workers for making thumbnails, if it only computes and the server has 8 cores?

Made by Ivo Gergov as part of Evo Tech's campaign for accessible programming education ·