Skip to content

← Concurrency with Threads and asyncio step 15 of 56

Medium Primitives

Write a Pool Map That Reports Every Failure

In the health check exercise, you sorted outcomes into dictionaries keyed by URL. Now write the general tool, a map for thread pools that keeps going when items fail. It takes any function and any items. It returns the results in input order and reports each failure with the position of the item that caused it.

values, failures = map_with_failures(int, ["10", "bad", "10"], max_workers=2)
# values   == [10, 10]
# failures == [(1, ValueError("invalid literal for int() with base 10: 'bad'"))]

What to write

def map_with_failures[T, R](
    fn: Callable[[T], R], items: Sequence[T], *, max_workers: int
) -> tuple[list[R], list[tuple[int, BaseException]]]:

Run fn on every item in a ThreadPoolExecutor with max_workers threads, and return (values, failures):

  • values holds the successful results in input order. Skip the items that failed.
  • failures holds one (index, exception) pair for each failed item, sorted by index. index is the item's position in items, and exception is the original exception object the job raised.

Follow these rules:

  • Submit each item once. Equal items are separate jobs, each with its own index.
  • Collect every item's outcome, and record each failure once. One failure must not stop the others.
  • Record every kind of exception, including BaseException subclasses such as KeyboardInterrupt. The driver raises one, and except Exception around result() would let it escape.
  • Shut the pool down and wait for its threads before you return. A with block does both.
  • Order both lists by index. The jobs can finish in a different order on each run.

Edge cases

  • Empty items returns ([], []).
  • A max_workers of 0 or less raises ValueError, even when items is empty.

What the driver checks

Keep solve as it is. It passes the items 0, 1, ..., len(kinds) - 1. The job for item i sleeps briefly, and later items sleep less, so they often finish first. Then the job returns i * 3 + 1 for "ok", raises ValueError for "boom", or raises KeyboardInterrupt for "base". The report has these keys:

  • values is your first list.
  • failures maps each failed index to "ExceptionName: message".
  • failure_order lists the indexes from your failure list, in your order. It must be increasing.
  • attempted counts how many times fn ran. It must equal the number of items.
  • thread_delta is the number of running threads after your call minus the number before it. It must be 0, which shows that no pool thread was left running.