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 Java this is ExecutorService, and each result comes back as a Future.

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:

List<Thread> threads = new ArrayList<>();
for (String image : uploads) {
    Thread thread = new Thread(() -> saveThumbnail(image));
    thread.start();
    threads.add(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 (usually up to 1 MB) and a place in the operating system. At a few thousand threads the JVM stops with OutOfMemoryError: unable to create native thread.
  • Even if there is memory, 10,000 threads on 8 cores only take turns. The processor wastes time switching them, and every image gets slower.

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: ExecutorService, submit and Future.get(). 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.
static int thumbnail(String image) throws InterruptedException {
    if (image.startsWith("broken")) {
        throw new IllegalArgumentException("corrupt file");
    }
    int kb = 20 + 3 * image.length();
    Thread.sleep(2 * kb);
    return kb;
}

Eight images, a pool of three workers:

try (var pool = Executors.newFixedThreadPool(3)) {
    List<Future<Integer>> sizes = new ArrayList<>();
    for (String image : images) {
        sizes.add(pool.submit(() -> thumbnail(image)));
    }
    int total = 0;
    for (int i = 0; i < images.size(); i++) {
        int kb = sizes.get(i).get();
        System.out.printf("%s: %d KB%n", images.get(i), kb);
        total += kb;
    }
    System.out.println("Total: " + total + " KB");
}
Step by step, in words
  1. 8 images wait. The pool is 3 threads: Executors.newFixedThreadPool(3).
  2. submit 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. get() reads the sizes in the order of the images, not in the order they finished. Total 379 KB; the end of try closes the pool.
List<String> images = List.of("boots.jpg", "jacket.jpg", "hat.jpg", "scarf.jpg", "backpack.jpg", "gloves.jpg", "watch.jpg", "cap.jpg");
try (var pool = Executors.newFixedThreadPool(3)) {
    List<Future<Integer>> sizes = new ArrayList<>();
    for (String image : images) {
        sizes.add(pool.submit(() -> thumbnail(image)));
    }
    int total = 0;
    for (int i = 0; i < images.size(); i++) {
        int kb = sizes.get(i).get();
        System.out.printf("%s: %d KB%n", images.get(i), kb);
        total += kb;
    }
    System.out.println("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
  • Executors.newFixedThreadPool(3) makes three threads and an endless queue in front of them.
  • submit puts the task into the queue and gives back a Future right away. The loop does not wait for anything.
  • get() waits for exactly its own task. The loop waits for them in the order of the images, so the output is in that order too.
  • try (var pool = …) closes the pool at the end: close() takes no new tasks, waits for the started ones and stops the threads. This is from Java 19. In older code you will see shutdown() and awaitTermination in a finally.
More precisely, if you are curious

The queue of newFixedThreadPool is endless. If uploads always come faster than the pool can handle, the queue grows until the memory runs out. A pool with a bounded queue is new ThreadPoolExecutor(3, 3, 0, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1000), new ThreadPoolExecutor.CallerRunsPolicy()): when the queue is full, the task runs in the thread that sends it, so accepting uploads only slows down.

How it is in the other languages: in C# there is one shared pool (ThreadPool, Task.Run, Parallel), in Python it is ThreadPoolExecutor, and Node has no ready-made pool of Workers. In a project there you use the Piscina package.

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:

long start = System.nanoTime();
List<Integer> sizes = new ArrayList<>();
try (var pool = Executors.newFixedThreadPool(3)) {
    for (String image : images) {
        Future<Integer> size = pool.submit(() -> thumbnail(image));
        sizes.add(size.get());
    }
}
long ms = (System.nanoTime() - start) / 1_000_000;
System.out.println("Took over 0.5 s: " + (ms > 500));
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. ExecutorCompletionService gives them that way. Then you only collect things that do not depend on the order:

try (var pool = Executors.newFixedThreadPool(3)) {
    var done = new ExecutorCompletionService<Integer>(pool);
    for (String image : images) {
        done.submit(() -> thumbnail(image));
    }
    int count = 0;
    int total = 0;
    for (int i = 0; i < images.size(); i++) {
        total += done.take().get();
        count++;
    }
    System.out.println("Done: " + count + " of " + images.size() + ", total: " + total + " KB");
}
Output
Done: 8 of 8, total: 379 KB
ExecutorCompletionService keeps its own queue of finished tasks. take() waits for the next finished one, whichever it is.

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:

List<String> images = List.of("boots.jpg", "jacket.jpg", "broken.jpg", "hat.jpg");
try (var pool = Executors.newFixedThreadPool(3)) {
    List<Future<Integer>> sizes = new ArrayList<>();
    for (String image : images) {
        sizes.add(pool.submit(() -> thumbnail(image)));
    }
    for (int i = 0; i < images.size(); i++) {
        try {
            int kb = sizes.get(i).get();
            System.out.println(images.get(i) + ": " + kb + " KB");
        } catch (ExecutionException e) {
            String why = e.getCause().getMessage();
            System.out.println(images.get(i) + ": error: " + why);
        }
    }
}
Output
boots.jpg: 47 KB
jacket.jpg: 50 KB
broken.jpg: error: corrupt file
hat.jpg: 41 KB
get() throws an ExecutionException, and the real exception is in getCause(). The other images do not suffer: the pool goes on with the next tasks.

Try it

Where did the error go

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

List<String> images = List.of("boots.jpg", "broken.jpg", "hat.jpg");
try (var pool = Executors.newFixedThreadPool(3)) {
    for (String image : images) {
        pool.submit(() -> thumbnail(image));
    }
}
System.out.println("Done: " + images.size() + " 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.
static int poolSize(int cores, int waitMs, int computeMs) {
    return cores * (1 + waitMs / computeMs);
}


System.out.println("Thumbnails (computing only): " + poolSize(8, 0, 50));
System.out.println("E-mails (90 ms waiting, 10 ms computing): " + poolSize(8, 90, 10));
Output
Thumbnails (computing only): 8
E-mails (90 ms waiting, 10 ms computing): 80

The number of cores comes from Runtime.getRuntime().availableProcessors(). 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.

Try it

Fill in

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

static int poolSize(int cores, int waitMs, int computeMs) {
    return cores * (1 + );
}
More precisely, if you are curious

Virtual threads (Java 21) change the math for work that waits. A virtual thread costs a few hundred bytes, and when it waits (sleep, network, database), the JVM takes it off the real thread and runs another one. Executors.newVirtualThreadPerTaskExecutor() gives a new virtual thread for every task, and 10,000 e-mails really are 10,000 threads, without a pool. For work that computes they do not help: there are still 8 cores. And the resources behind them (database connections) are still limited, for example with a Semaphore.

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? (ArrayBlockingQueue and CallerRunsPolicy)
  • 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 ·