Python Task Race Lab

Compare Python threads and processes in a tested concurrency benchmark.

Introduction

30 Second Summary

Some tasks keep your computer busy while others spend most of their time waiting. Without measurements, both can look slow for completely different reasons.

In this project, you will build a terminal experiment in Python that compares sequential execution, threads, and processes across waiting tasks and prime checks. The terminal report makes each strategy's behavior visible.

What You'll Build

Running myscript.py fills your terminal with measured races, named workers, matching result checks, and a passing test report.

By the end of this project, you'll have:

  • A wait benchmark where you can compare one-at-a-time execution with overlapping work from four threads.
  • A CPU benchmark where child processes return prime-check results with their process IDs.
  • A pickling failure you can trigger, diagnose, and repair before watching three correctness tests pass.
  • Secret Mission: Build a workload selector that maps io to threaded waits. It maps cpu to process-based prime checks. Two extra tests verify valid choices and rejected workloads.

Do I need anything before starting?

You need Python 3.14 plus Visual Studio Code on macOS.

Basic Python syntax and functions are enough for this standard-library-only lab.

Before We Start

This is your moment to commit to the question behind the lab before any hands-on work begins. You will test your predictions about sequential execution, threads, processes, and the GIL using measurable evidence.

Prepare the Lab

Your experiment needs a controlled workspace before you can test sequential execution, threads, processes, or the Global Interpreter Lock with measurable evidence.

A controlled Visual Studio Code workspace keeps editor confusion out of your experiment. A verified Python 3.14 interpreter keeps version differences out of your results.

In this step, get ready to:
  • Create the python-task-race-lab folder on your Desktop.
  • Confirm the integrated terminal reports Python 3.14.
  • Create an empty myscript.py file.
Open the lab workspace

The workspace folder gives every part of the experiment one known location. Keeping it on your Desktop makes it easy to find during each benchmark.

  • Click the Desktop to make Finder active.
  • Click File in the menu bar.
  • Click New Folder.
  • Enter python-task-race-lab as the folder name.
  • Press Return to save the name.

You'll see the python-task-race-lab folder on your Desktop.

  • Press Command-Space bar to open Spotlight.
  • Enter Visual Studio Code in the search field.
  • Press Return to open the selected app.

✔️ Visual Studio Code opens

Visual Studio Code is ready to hold the lab workspace.

ⓧ Visual Studio Code is missing

Installing an app from a disk image can feel fiddly the first time. The official macOS package places Visual Studio Code in your Applications folder.

  • Open the official Visual Studio Code installation guide in your browser.
  • Download Visual Studio Code for macOS.
  • Open the downloaded .dmg file.
  • Drag Visual Studio Code.app to the Applications folder.
  • Press Command-Space bar to reopen Spotlight.
  • Enter Visual Studio Code in the search field.
  • Press Return to open the installed app.
  • Click File in the Visual Studio Code menu bar.
  • Click Open Folder.
  • Select the python-task-race-lab folder on your Desktop.
  • Click Open.
  • Confirm that you trust the folder if Visual Studio Code asks.

You'll see python-task-race-lab as the active folder in the Explorer sidebar. Your experiment now has a fixed home.

Folder Missing from the Explorer?

Check that you selected python-task-race-lab itself when using Open Folder. Selecting the Desktop would open every Desktop item as part of the workspace.

Use this prompt for focused help: help me open my Desktop folder named python-task-race-lab in Visual Studio Code on macOS.

Verify the Python interpreter

The benchmarks rely on behavior available in the Python 3.14 series. The integrated terminal shows which interpreter this workspace actually uses.

  • Click View in the Visual Studio Code menu bar.
  • Click Terminal.

You'll see the integrated terminal beneath the editor area.

Before you run the check, do you expect the terminal to report the required Python series?

  • Verify the active Python interpreter by running this command:
python3 --version

What Does This Check Prove?

The python3 command reports the interpreter selected by the integrated terminal. Output beginning with Python 3.14 confirms that the workspace uses the required Python series.

✔️ I see the required version

That's the interpreter confirmed. Every benchmark in this project can now run against the intended Python series.

ⓧ I see an older version

Your terminal is resolving an older Python installation. Installing the required series gives the lab access to the interpreter behavior used by the finished experiment.

  • Open the official Python macOS installation guide in your browser.
  • Install the Python 3.14 macOS release by following the official guide.
  • Restart Visual Studio Code after the installation finishes.
  • Click View in the Visual Studio Code menu bar.
  • Click Terminal.
  • Repeat the version check shown above.

ⓧ Command not found

The terminal cannot find Python on its command path. The official macOS installation guide provides the required interpreter.

  • Open the official Python macOS installation guide in your browser.
  • Install the Python 3.14 macOS release by following the official guide.
  • Restart Visual Studio Code after the installation finishes.
  • Click View in the Visual Studio Code menu bar.
  • Click Terminal.
  • Repeat the version check shown above.

Still Seeing the Wrong Python Version?

A terminal that was already open can keep an older command path. Restart Visual Studio Code before repeating the check.

Share the output with the helper prompt: help me make the Visual Studio Code terminal report Python 3.14 on macOS.

Create the empty script

The entire experiment lives in one file named myscript.py. Starting with a blank file keeps the first benchmark easy to trace.

  • Click the New File button in the Explorer toolbar.
  • Enter myscript.py as the filename.
  • Press Return to create the file.

You'll see an empty myscript.py editor tab inside the python-task-race-lab workspace.

Before the final check, do you expect the terminal to report the same interpreter after creating the script?

  • Confirm the workspace still uses the required interpreter by running this command:
python3 --version

What Does the Final Check Confirm?

Running the check from the integrated terminal confirms that your open workspace can access Python. Output beginning with Python 3.14 is the success signal for this step.

You'll see output beginning with Python 3.14. The Explorer also lists the blank myscript.py file.

✔️ Awesome, I've got everything!

Your Explorer lists myscript.py inside python-task-race-lab. The file is blank and ready for the first benchmark.

ⓧ I'd like to double check the full code

The exact cumulative content of myscript.py is a blank file. There should be no characters or code in the editor.

  • Remove any characters from myscript.py.
  • Save the empty file.

Your lab is ready for evidence-driven experiments. Next, you'll measure four waiting tasks running one at a time.

Measure Sequential Waits

Your Visual Studio Code workspace gives this experiment a controlled home. The first question is how long four waiting tasks take when Python handles them one at a time.

Sequential execution creates the baseline that every later runner must beat. Each task pauses for 0.3 seconds.

Measuring the combined run makes the waiting bottleneck visible. That evidence gives you something concrete to compare with threads in the next step.

In this step, get ready to:
  • Create a wait task that reports its duration plus its thread identity plus its process identity.
  • Measure four sequential waits with elapsed wall-clock time.
  • Run the guarded script to establish your thread prediction.
Model one waiting task

A waiting task models work that spends time paused. Its returned record identifies the duration plus the worker plus the operating-system process.

  • Return to the empty myscript.py file in the Visual Studio Code editor.
  • Add the imports plus the wait durations plus the simulate_wait() function by pasting this code:
import os
import threading
import time


WAIT_DURATIONS = [0.3, 0.3, 0.3, 0.3]


def simulate_wait(duration):
    time.sleep(duration)
    return {
        "duration": duration,
        "thread": threading.current_thread().name,
        "pid": os.getpid(),
    }

How Does One Wait Become Evidence?

  • The imports provide the clock plus the worker name plus the operating-system identifier used by each result.
  • WAIT_DURATIONS supplies four identical 0.3-second tasks.
  • time.sleep() pauses the current worker for the supplied duration.
  • threading.current_thread().name records the worker name.
  • os.getpid() records the process identifier.
  • simulate_wait() returns these observations as a dictionary that later runners can compare.
Add the sequential timer

Elapsed wall-clock time captures the complete run from start to finish. Keeping the timer separate from the runner lets later execution strategies use the same measurement method.

  • Paste this code below simulate_wait() in myscript.py:
def timed_run(label, runner):
    started = time.perf_counter()
    results = runner()
    elapsed = time.perf_counter() - started
    print(f"{label}: {elapsed:.2f} seconds")
    return results, elapsed


def run_sequential_waits():
    return [simulate_wait(duration) for duration in WAIT_DURATIONS]


def main():
    timed_run("Sequential waits", run_sequential_waits)


if __name__ == "__main__":
    main()

What Does This Code Do?

  • timed_run() records a high-resolution timestamp before the runner starts. It subtracts that value from a second timestamp after the runner finishes.
  • run_sequential_waits() calls simulate_wait() once for each value in WAIT_DURATIONS.
  • main() passes the sequential runner into the shared timer.
  • if __name__ == "__main__": runs main() when you execute this file as a script.
  • Save myscript.py.
Run the sequential baseline

The timer now wraps all four waits as one experiment. The printed duration shows the cost of handling those pauses one at a time.

Before you run the script, do you expect the measured total to be closer to 0.3 seconds or 1.2 seconds?

  • Execute the sequential wait benchmark by running this command in the integrated terminal:
python3.14 myscript.py

What Should You See?

You should see one line beginning with Sequential waits: followed by a measured duration in seconds.

A result near 1.20 seconds reflects four 0.3-second waits running one after another. Your measured value can be slightly higher because elapsed timing depends on the machine.

That is your first piece of evidence. Four waits have become one measured sequential baseline.

No Sequential Result?

Confirm that myscript.py is saved before rerunning the command.

Check that main() is aligned at module scope. Check that the final main() call is indented beneath if __name__ == "__main__":.

Help me debug my sequential wait script.

✔️ Awesome, I've got everything!

Great. Confirm that myscript.py is saved with the guarded entry point in place.

ⓧ I'd like to double check the full code

Compare your saved myscript.py file with this complete version.

import os
import threading
import time


WAIT_DURATIONS = [0.3, 0.3, 0.3, 0.3]


def simulate_wait(duration):
    time.sleep(duration)
    return {
        "duration": duration,
        "thread": threading.current_thread().name,
        "pid": os.getpid(),
    }


def timed_run(label, runner):
    started = time.perf_counter()
    results = runner()
    elapsed = time.perf_counter() - started
    print(f"{label}: {elapsed:.2f} seconds")
    return results, elapsed


def run_sequential_waits():
    return [simulate_wait(duration) for duration in WAIT_DURATIONS]


def main():
    timed_run("Sequential waits", run_sequential_waits)


if __name__ == "__main__":
    main()

Your sequential baseline is measured. Next, you will reuse the same waiting task with worker threads to see whether overlapping idle time changes the result.

Overlap Waits with Threads

Your sequential baseline makes the waiting bottleneck visible. Each task sleeps while the next task waits for its turn.

A thread pool can overlap those pauses while reusing simulate_wait(). This gives you direct evidence of how threads help waiting-heavy work.

In this step, get ready to:
  • Build a four-worker thread runner for the existing wait tasks.
  • Measure the sequential wait runner beside the threaded wait runner.
  • Compare returned durations for correctness.
Build the threaded wait runner

The concurrent.futures module provides a shared executor interface for thread pools and process pools. Its mapping interface lets several workers apply the same function to the same collection of inputs.

Why Use the Executor Interface?

The shared interface keeps this experiment focused on workload choice. You can compare execution strategies without learning unrelated pool APIs.

  • At the top of myscript.py find this first import:
import os

Where Does the Import Go?

The new module belongs at the start of the standard-library import group. Keeping imports together makes each dependency easy to locate.

  • Replace that line with this import group:
import concurrent.futures
import os

What Does This Import Add?

The concurrent.futures module provides ThreadPoolExecutor. Your existing imports continue to provide process IDs, thread names, and timing.

  • Scroll below run_sequential_waits() in myscript.py.
  • Add the threaded runner with this code:
def run_threaded_waits():
    with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
        return list(executor.map(simulate_wait, WAIT_DURATIONS))

What Does This Code Do?

  • The ThreadPoolExecutor(max_workers=4) context creates up to four worker threads.
  • The executor.map call applies simulate_wait to every value in WAIT_DURATIONS.
  • The context manager waits for pending work before closing the pool.
  • The returned list preserves the input order for comparison with the sequential results.
  • Save myscript.py.
  • Confirm the sequential baseline still runs by executing:
python3.14 myscript.py

What Should You See?

You should still see the Sequential waits measurement. This proves the new runner has not disrupted the existing path.

Good work. Your original benchmark still runs with the threaded path ready beside it.

Seeing an Import or Syntax Error?

Check that import concurrent.futures starts at the left edge of the file. Confirm that run_threaded_waits() sits outside every other function.

Ask for help with the exact file structure through this threading setup check.

Time both wait modes

Both runners now perform the same four wait tasks. Measuring them through timed_run() keeps task results separate from elapsed wall-clock time.

  • Select the current main() function in myscript.py.
  • Replace that function with this version:
def main():
    sequential_waits, sequential_wait_time = timed_run(
        "Sequential waits",
        run_sequential_waits,
    )
    threaded_waits, threaded_wait_time = timed_run(
        "Threaded waits",
        run_threaded_waits,
    )

What Does This Code Measure?

  • The first call collects the sequential task records plus their elapsed time.
  • The second call collects the threaded task records plus their elapsed time.
  • Each result set remains available for the correctness comparison.
  • Save myscript.py.

Before you run the comparison, do you expect four threads to reduce the time spent waiting?

  • Run both wait modes by executing:
python3.14 myscript.py

What Should You See?

You should see a Sequential waits measurement followed by a Threaded waits measurement. The threaded measurement normally reports a lower elapsed time because the sleeps overlap.

Does the Threaded Measurement Fail?

Check the spelling of run_threaded_waits in both its definition and the timed_run() call. Confirm that ThreadPoolExecutor uses the imported concurrent.futures module.

Use this comparison debugging prompt if the threaded line does not print.

Compare results and worker identities

Elapsed time shows what happened during one benchmark run. Returned durations provide the stable correctness evidence because both strategies receive the same inputs.

  • Place this reporting code inside main() after the threaded timed_run() call:
    print(
        "Sequential threads:",
        sorted({result["thread"] for result in sequential_waits}),
    )
    print(
        "Thread-pool workers:",
        sorted({result["thread"] for result in threaded_waits}),
    )
    print(
        "Wait results match:",
        [result["duration"] for result in sequential_waits]
        == [result["duration"] for result in threaded_waits],
    )
    print(
        "Observed wait-time improvement:",
        sequential_wait_time - threaded_wait_time,
    )

What Does This Report Prove?

  • The sequential report identifies the thread that handled every baseline task.
  • The thread-pool report reveals the worker thread names used for overlapping waits.
  • The duration comparison checks that both strategies returned equivalent task results.
  • The final subtraction records the observed timing difference from this run.

Why Keep Timing Out of Tests?

Elapsed time varies with machine load and operating-system scheduling. Correctness tests should compare returned values so the result stays dependable across different computers.

  • Save myscript.py.

Before you run the complete comparison, which values do you expect to stay equal? Which thread identities do you expect to change?

  • Run the completed wait comparison by executing:
python3.14 myscript.py

What Should You See?

  • You should see measurements for Sequential waits and Threaded waits.
  • You should see one sequential thread name plus thread-pool worker names.
  • You should see Wait results match: True.
  • You should normally see a positive number beside Observed wait-time improvement:.

That comparison is working. Your lab now shows overlapping waits while preserving the same task results.

Missing Worker Names or a Matching Result?

Confirm that both runners use the same WAIT_DURATIONS list. Check that executor.map receives simulate_wait as its function.

Use this result comparison check to trace a mismatch.

✔️ Awesome, I've got everything!

Your sequential and threaded wait measurements are ready. Make sure you have saved myscript.py.

ⓧ I'd like to double check the full code

Compare your saved file with this complete version.

import concurrent.futures
import os
import threading
import time


WAIT_DURATIONS = [0.3, 0.3, 0.3, 0.3]


def simulate_wait(duration):
    time.sleep(duration)
    return {
        "duration": duration,
        "thread": threading.current_thread().name,
        "pid": os.getpid(),
    }


def timed_run(label, runner):
    started = time.perf_counter()
    results = runner()
    elapsed = time.perf_counter() - started
    print(f"{label}: {elapsed:.2f} seconds")
    return results, elapsed


def run_sequential_waits():
    return [simulate_wait(duration) for duration in WAIT_DURATIONS]


def run_threaded_waits():
    with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
        return list(executor.map(simulate_wait, WAIT_DURATIONS))


def main():
    sequential_waits, sequential_wait_time = timed_run(
        "Sequential waits",
        run_sequential_waits,
    )
    threaded_waits, threaded_wait_time = timed_run(
        "Threaded waits",
        run_threaded_waits,
    )

    print(
        "Sequential threads:",
        sorted({result["thread"] for result in sequential_waits}),
    )
    print(
        "Thread-pool workers:",
        sorted({result["thread"] for result in threaded_waits}),
    )
    print(
        "Wait results match:",
        [result["duration"] for result in sequential_waits]
        == [result["duration"] for result in threaded_waits],
    )
    print(
        "Observed wait-time improvement:",
        sequential_wait_time - threaded_wait_time,
    )


if __name__ == "__main__":
    main()

How to Use This Reference

This file is the complete checkpoint for this step. Compare the import order, function indentation, and reporting lines with your saved version.

Your wait benchmark now measures sequential work beside overlapping thread work. Next, you will test CPU-heavy work and encounter the boundary between threads and separate processes.

Hit the Process Boundary

Your thread pool now overlaps waiting tasks. The returned durations still match the sequential baseline.

A process pool sends work to separate Python interpreters. This model gives CPU-bound work a path beyond one interpreter.

Separate interpreters introduce a serialization boundary. This experiment reveals that boundary through a worker defined inside another function.

In this step, get ready to:
  • Build a sequential prime-checking baseline.
  • Submit the same prime checks to separate processes.
  • Inspect the boundary revealed by a local worker function.
Build the sequential prime baseline

Prime checking keeps the processor busy with repeated calculations. A sequential run gives the process experiment a correctness baseline.

  • In the myscript.py tab from earlier, replace the opening imports through WAIT_DURATIONS with this block:
import concurrent.futures
import math
import os
import threading
import time


WAIT_DURATIONS = [0.3, 0.3, 0.3, 0.3]

PRIME_CANDIDATES = [
    112272535095293,
    112582705942171,
    112272535095293,
    115280095190773,
    115797848077099,
    1099726899285419,
]

What does this setup add?

  • The math module supplies the square-root calculation used to limit each prime search.
  • PRIME_CANDIDATES stores the six numbers used by both CPU runners.
  • Using one candidate list keeps the sequential result comparable with the process result.
  • In myscript.py, add is_prime() below simulate_wait() by copying this function:
def is_prime(number):
    if number < 2:
        return False
    if number == 2:
        return True
    if number % 2 == 0:
        return False

    limit = int(math.floor(math.sqrt(number)))
    for divisor in range(3, limit + 1, 2):
        if number % divisor == 0:
            return False
    return True

How does the prime check work?

  • Values below 2 cannot be prime.
  • The function handles 2 before rejecting other even numbers.
  • The loop tests odd divisors up to the square root of the candidate.
  • Finding one divisor returns False immediately.
  • Save myscript.py.
  • Check that Python can parse the expanded file by running this command in the integrated terminal:
python3.14 myscript.py

What does this checkpoint prove?

Python parses the new import data before running the existing wait comparison. Reaching the wait-time report confirms that the prime-checking foundation has valid syntax.

You should see the sequential wait result followed by the threaded wait result. The thread names and duration comparison should still print.

Seeing a syntax error?

Check that PRIME_CANDIDATES closes with a square bracket. Check that is_prime() starts at the left edge of the file.

Ask for help with the new prime-checking code.

Run the sequential CPU benchmark

The prime predicate can now evaluate one number. A runner applies it to every candidate in the original process.

  • In myscript.py, add run_sequential_primes() below run_threaded_waits() by copying this function:
def run_sequential_primes():
    return [
        (number, is_prime(number))
        for number in PRIME_CANDIDATES
    ]

What does this runner return?

The list comprehension calls is_prime() once for each candidate. Every result pairs the original number with its Boolean prime result.

  • At the end of main(), locate the closing print block for Observed wait-time improvement:.
  • Add this line immediately below that print block:
    timed_run("Sequential prime checks", run_sequential_primes)

Why use timed_run again?

The existing timed_run() helper keeps measurement separate from prime-checking logic. The label makes the new CPU baseline visible in the terminal.

  • Save myscript.py.

Before you run the file, where do you expect the sequential prime timing to appear relative to the wait comparison?

  • Run the expanded benchmark by using this command in the integrated terminal:
python3.14 myscript.py

What changed in the report?

The complete wait comparison prints first. A new line beginning with Sequential prime checks appears after it.

That new timing proves every candidate completed in the main process. It becomes the baseline for the process-pool attempt.

That is the CPU baseline locked in. Your lab now measures correct prime work before changing its execution model.

Missing the prime timing?

Confirm that the timed_run() call sits inside main(). Its indentation should match the wait-report print blocks.

Ask for help wiring the sequential prime runner into main.

Send the prime work to processes

Process workers have separate memory spaces. ProcessPoolExecutor can move CPU-heavy Python work beyond the Global Interpreter Lock on a regular GIL-enabled build.

Every submitted callable must cross into those worker processes. For this experiment, the worker stays inside the process runner so you can test that boundary directly.

  • In myscript.py, add run_process_primes() below run_sequential_primes() by copying this function:
def run_process_primes():
    def prime_with_pid(number):
        return number, is_prime(number), os.getpid()

    with concurrent.futures.ProcessPoolExecutor(max_workers=2) as executor:
        return list(executor.map(prime_with_pid, PRIME_CANDIDATES))

What does this process runner attempt?

  • prime_with_pid() returns the candidate number.
  • The same tuple includes the prime result.
  • The process ID records which process handled the candidate.
  • ProcessPoolExecutor(max_workers=2) creates a pool with at most two worker processes.
  • executor.map() attempts to submit the local worker for every candidate.
  • In main(), locate the sequential prime timing you added.
  • Add this process-pool timing immediately below it:
    timed_run("Process-pool prime checks", run_process_primes)

Where does this call run?

The call remains inside the protected main() path. It starts only after the working wait comparison and sequential prime baseline finish.

  • Save myscript.py.

Before you run this experiment, do you think a child process can receive prime_with_pid() from inside run_process_primes()?

  • Test the process boundary by running this command in the integrated terminal:
python3.14 myscript.py

The Process Boundary Is Visible

The wait comparison completes first. The sequential prime timing also completes.

The process-pool attempt stops with a traceback that mentions pickling. This failure is the intended result of the experiment.

The traceback identifies the local prime_with_pid() function. A child process cannot import that function from module scope because it is nested inside run_process_primes().

  • Scroll through the traceback to locate the message that mentions pickling.
  • Trace the named callable back to the local prime_with_pid() definition.

You have reproduced the serialization boundary directly. Threads could call the existing task function in shared memory, while process workers need an importable callable.

Did the traceback appear too early?

The wait comparison and sequential prime timing should complete before the process-pool traceback. An earlier syntax error usually points to indentation around run_process_primes().

Ask for help locating an unexpected earlier failure.

✔️ I reached the pickling boundary

Your working wait comparison reaches the sequential prime timing before the local process worker triggers the intended traceback.

ⓧ I'd like to double check the full code

Compare your complete myscript.py file with this version. The local worker is intentionally preserved for the boundary experiment.

import concurrent.futures
import math
import os
import threading
import time


WAIT_DURATIONS = [0.3, 0.3, 0.3, 0.3]

PRIME_CANDIDATES = [
    112272535095293,
    112582705942171,
    112272535095293,
    115280095190773,
    115797848077099,
    1099726899285419,
]


def simulate_wait(duration):
    time.sleep(duration)
    return {
        "duration": duration,
        "thread": threading.current_thread().name,
        "pid": os.getpid(),
    }


def is_prime(number):
    if number < 2:
        return False
    if number == 2:
        return True
    if number % 2 == 0:
        return False

    limit = int(math.floor(math.sqrt(number)))
    for divisor in range(3, limit + 1, 2):
        if number % divisor == 0:
            return False
    return True


def timed_run(label, runner):
    started = time.perf_counter()
    results = runner()
    elapsed = time.perf_counter() - started
    print(f"{label}: {elapsed:.2f} seconds")
    return results, elapsed


def run_sequential_waits():
    return [simulate_wait(duration) for duration in WAIT_DURATIONS]


def run_threaded_waits():
    with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
        return list(executor.map(simulate_wait, WAIT_DURATIONS))


def run_sequential_primes():
    return [
        (number, is_prime(number))
        for number in PRIME_CANDIDATES
    ]


def run_process_primes():
    def prime_with_pid(number):
        return number, is_prime(number), os.getpid()

    with concurrent.futures.ProcessPoolExecutor(max_workers=2) as executor:
        return list(executor.map(prime_with_pid, PRIME_CANDIDATES))


def main():
    sequential_waits, sequential_wait_time = timed_run(
        "Sequential waits",
        run_sequential_waits,
    )
    threaded_waits, threaded_wait_time = timed_run(
        "Threaded waits",
        run_threaded_waits,
    )

    print(
        "Sequential threads:",
        sorted({result["thread"] for result in sequential_waits}),
    )
    print(
        "Thread-pool workers:",
        sorted({result["thread"] for result in threaded_waits}),
    )
    print(
        "Wait results match:",
        [result["duration"] for result in sequential_waits]
        == [result["duration"] for result in threaded_waits],
    )
    print(
        "Observed wait-time improvement:",
        sequential_wait_time - threaded_wait_time,
    )

    timed_run("Sequential prime checks", run_sequential_primes)
    timed_run("Process-pool prime checks", run_process_primes)


if __name__ == "__main__":
    main()

You have made the process boundary visible with a real failure. Next, you will move the worker to module scope and prove the repaired process pool with automated tests.

Fix and Test the Process Pool

Your wait comparison works. The process benchmark has exposed the serialization boundary between the main program and its child processes.

Child processes need an importable worker. This step moves the worker to module scope. You will also add correctness-focused tests that avoid hardware-specific timing thresholds.

In this step, get ready to:
  • Move prime_with_pid() to module scope so child processes can import it.
  • Report the Global Interpreter Lock status plus the main and worker process IDs.
  • Add three unit tests that verify results without requiring a specific speedup.
Move the worker to module scope

The local prime_with_pid() function caused the earlier pickling failure. A module-scope function has a stable import path that each process worker can resolve.

  • In myscript.py, locate run_process_primes() below run_sequential_primes().
  • Delete the nested prime_with_pid() definition from run_process_primes().
  • Add the module-scope worker below is_prime() by copying this code:
def prime_with_pid(number):
    return number, is_prime(number), os.getpid()

Why Does This Fix the Worker?

  • The module-scope location lets each child process import prime_with_pid() by name.
  • The returned tuple keeps the candidate number with its prime result.
  • The value from os.getpid() provides evidence that a child process completed the work.
  • Save myscript.py.
  • Keep run_process_primes() mapped to the module-scope prime_with_pid() function.

Before you run the script, do you expect the process-pool traceback to return?

  • Run the corrected benchmark in the integrated terminal by running this command:
python3.14 myscript.py

What Should You See?

The wait comparison completes first. The script then prints a duration for Process-pool prime checks without the pickling-related traceback.

That boundary is fixed. Your child processes can now receive the worker function and return prime results.

Still Seeing a Pickling-Related Failure?

Check that prime_with_pid() starts at the far left of the file. Any indentation beneath run_process_primes() keeps it local.

Confirm that executor.map() still receives prime_with_pid without parentheses.

Help me check the worker location and process-pool mapping.

Report process identities

A successful result proves the process pool runs. Process IDs make the execution model visible by showing that the prime checks happen outside the main process.

  • Replace the import group at the top of myscript.py with this complete group:
import concurrent.futures
import math
import os
import sys
import threading
import time
import unittest

What Do the New Imports Support?

  • The sys module exposes whether this CPython build has the GIL enabled.
  • The unittest module provides the test case and test runner used later in this step.
  • In main(), add the lab title plus the GIL report above the sequential wait benchmark:
def main():
    print("Python Task Race Lab")
    print(f"GIL enabled: {sys._is_gil_enabled()}")

What Does the GIL Report Show?

The value from sys._is_gil_enabled() identifies the interpreter mode used for this run. The thread guidance in this lab applies to a regular GIL-enabled CPython build.

  • Save myscript.py.
  • Run the updated report in the integrated terminal by running this command:
python3.14 myscript.py

Check the Interpreter Report

You should see Python Task Race Lab followed by a line beginning with GIL enabled:.

  • At the end of main(), replace the two existing prime timing calls with this CPU report:
    sequential_primes, _ = timed_run(
        "Sequential prime checks",
        run_sequential_primes,
    )
    process_primes, _ = timed_run(
        "Process-pool prime checks",
        run_process_primes,
    )
    process_values = [
        (number, prime)
        for number, prime, _ in process_primes
    ]

    print("Main PID:", os.getpid())
    print(
        "Worker PIDs:",
        sorted({pid for _, _, pid in process_primes}),
    )
    print("Prime results match:", process_values == sequential_primes)

How Does the CPU Report Prove the Result?

  • The sequential runner supplies the expected number and prime pairs.
  • The process runner supplies the same pairs plus the process ID for each result.
  • The list comprehension removes the process IDs before the correctness comparison.
  • Different main and worker PIDs prove that separate processes handled the mapped work.

Before you run the expanded report, do you expect the main PID to appear in the worker PID list?

  • Save myscript.py.
  • Run the process identity report in the integrated terminal by running this command:
python3.14 myscript.py

What Proves the Process Pool Works?

You should see a main PID plus one or more different worker PIDs. You should also see Prime results match: True.

Do the Prime Results Stop Early?

Check the indentation of the CPU report. Every line in the report belongs inside main().

Confirm that process_values keeps the number and prime result while dropping only the PID.

Help me debug the CPU report and PID comparison.

Add correctness-focused tests

Benchmark times vary with hardware and background activity. These tests check returned values plus process identity so correctness stays stable across machines.

  • At the bottom of myscript.py, locate the protected __main__ block.
  • Add the test runner beneath main() so the block looks like this:
if __name__ == "__main__":
    main()
    unittest.main(verbosity=2)

Why Keep Startup Protected?

The protected block runs the benchmark plus the tests only when you execute myscript.py directly. Child processes can import the module without starting another copy of the entire program.

  • Above main(), add the test class with its wait-result test:
class TaskRaceTests(unittest.TestCase):
    def test_wait_modes_return_same_durations(self):
        sequential = [
            result["duration"]
            for result in run_sequential_waits()
        ]
        threaded = [
            result["duration"]
            for result in run_threaded_waits()
        ]
        self.assertEqual(sequential, threaded)

What Does the Wait Test Prove?

The test extracts only the returned durations from both runners. assertEqual() verifies that threading preserves the result sequence without asserting a particular elapsed time.

  • Save myscript.py.
  • Run the first automated check in the integrated terminal by running this command:
python3.14 myscript.py

Check the First Test

The benchmark report completes before one test runs. The test report ends with OK.

Does the Wait Test Fail?

Confirm that both comprehensions read the duration key. Thread names and process IDs differ by execution mode, so they do not belong in this equality check.

Help me compare the sequential and threaded wait results.

  • Inside TaskRaceTests, add the known-value prime test below the wait test:
    def test_is_prime_with_known_values(self):
        cases = [
            (1, False),
            (2, True),
            (3, True),
            (4, False),
            (29, True),
        ]
        for number, expected in cases:
            with self.subTest(number=number):
                self.assertEqual(is_prime(number), expected)

Why Use Subtests?

Each candidate runs as a named subtest. A failure identifies the specific number while the remaining cases continue to run.

  • Save myscript.py.
  • Run both automated checks in the integrated terminal by running this command:
python3.14 myscript.py

Check the Prime Cases

The test report now runs two tests. A successful report ends with OK.

  • Inside TaskRaceTests, add the process-result test below the known-value test:
    def test_process_pool_matches_sequential_results(self):
        expected = run_sequential_primes()
        process_results = run_process_primes()
        actual = [
            (number, prime)
            for number, prime, _ in process_results
        ]
        worker_pids = [pid for _, _, pid in process_results]

        self.assertEqual(actual, expected)
        self.assertTrue(all(pid != os.getpid() for pid in worker_pids))

What Does the Process Test Check?

  • The first assertion checks that the process pool returns the same prime decisions as sequential execution.
  • The second assertion checks that every recorded worker PID differs from the current main-process PID.
  • Neither assertion depends on a timing threshold.

Before the final run, do you expect all three tests to pass on a slower computer?

  • Save myscript.py.
  • Run the completed lab in the integrated terminal by running this command:
python3.14 myscript.py

Verify the Complete Lab

You should see the GIL status plus both wait timings. You should also see both CPU timings plus different main and worker PIDs.

The report shows Prime results match: True. It ends with Ran 3 tests followed by OK.

Do the Process Tests Re-run the Script?

Confirm that both main() and unittest.main() remain inside the protected __main__ block.

Check that TaskRaceTests appears above main() so the test runner can discover all three methods.

Help me debug the protected startup path and test discovery.

✔️ Awesome, I've got everything!

Your saved myscript.py now runs the complete benchmark and finishes with three passing tests.

ⓧ I'd like to double check the full code

Compare your saved file with this complete version.

import concurrent.futures
import math
import os
import sys
import threading
import time
import unittest


WAIT_DURATIONS = [0.3, 0.3, 0.3, 0.3]

PRIME_CANDIDATES = [
    112272535095293,
    112582705942171,
    112272535095293,
    115280095190773,
    115797848077099,
    1099726899285419,
]


def simulate_wait(duration):
    time.sleep(duration)
    return {
        "duration": duration,
        "thread": threading.current_thread().name,
        "pid": os.getpid(),
    }


def is_prime(number):
    if number < 2:
        return False
    if number == 2:
        return True
    if number % 2 == 0:
        return False

    limit = int(math.floor(math.sqrt(number)))
    for divisor in range(3, limit + 1, 2):
        if number % divisor == 0:
            return False
    return True


def prime_with_pid(number):
    return number, is_prime(number), os.getpid()


def timed_run(label, runner):
    started = time.perf_counter()
    results = runner()
    elapsed = time.perf_counter() - started
    print(f"{label}: {elapsed:.2f} seconds")
    return results, elapsed


def run_sequential_waits():
    return [simulate_wait(duration) for duration in WAIT_DURATIONS]


def run_threaded_waits():
    with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
        return list(executor.map(simulate_wait, WAIT_DURATIONS))


def run_sequential_primes():
    return [
        (number, is_prime(number))
        for number in PRIME_CANDIDATES
    ]


def run_process_primes():
    with concurrent.futures.ProcessPoolExecutor(max_workers=2) as executor:
        return list(executor.map(prime_with_pid, PRIME_CANDIDATES))


class TaskRaceTests(unittest.TestCase):
    def test_wait_modes_return_same_durations(self):
        sequential = [
            result["duration"]
            for result in run_sequential_waits()
        ]
        threaded = [
            result["duration"]
            for result in run_threaded_waits()
        ]
        self.assertEqual(sequential, threaded)

    def test_is_prime_with_known_values(self):
        cases = [
            (1, False),
            (2, True),
            (3, True),
            (4, False),
            (29, True),
        ]
        for number, expected in cases:
            with self.subTest(number=number):
                self.assertEqual(is_prime(number), expected)

    def test_process_pool_matches_sequential_results(self):
        expected = run_sequential_primes()
        process_results = run_process_primes()
        actual = [
            (number, prime)
            for number, prime, _ in process_results
        ]
        worker_pids = [pid for _, _, pid in process_results]

        self.assertEqual(actual, expected)
        self.assertTrue(all(pid != os.getpid() for pid in worker_pids))


def main():
    print("Python Task Race Lab")
    print(f"GIL enabled: {sys._is_gil_enabled()}")

    sequential_waits, sequential_wait_time = timed_run(
        "Sequential waits",
        run_sequential_waits,
    )
    threaded_waits, threaded_wait_time = timed_run(
        "Threaded waits",
        run_threaded_waits,
    )

    print(
        "Sequential threads:",
        sorted({result["thread"] for result in sequential_waits}),
    )
    print(
        "Thread-pool workers:",
        sorted({result["thread"] for result in threaded_waits}),
    )
    print(
        "Wait results match:",
        [result["duration"] for result in sequential_waits]
        == [result["duration"] for result in threaded_waits],
    )
    print(
        "Observed wait-time improvement:",
        sequential_wait_time - threaded_wait_time,
    )

    sequential_primes, _ = timed_run(
        "Sequential prime checks",
        run_sequential_primes,
    )
    process_primes, _ = timed_run(
        "Process-pool prime checks",
        run_process_primes,
    )
    process_values = [
        (number, prime)
        for number, prime, _ in process_primes
    ]

    print("Main PID:", os.getpid())
    print(
        "Worker PIDs:",
        sorted({pid for _, _, pid in process_primes}),
    )
    print("Prime results match:", process_values == sequential_primes)


if __name__ == "__main__":
    main()
    unittest.main(verbosity=2)

You have turned a process-pool failure into a tested concurrency lab. Threads now demonstrate overlapping waits. Processes now demonstrate separate CPU workers with equivalent results.

Secret mission

Build an Executor Selector

Your lab can run waiting work with threads or CPU-heavy work with processes. Add a decision layer that selects the appropriate runner from a workload label without repeating either benchmark.

Clean Up Your Resources

Clean Up Your Resources

Everything in this lab runs locally with no ongoing costs. Choose whether to keep, pause, or delete the workspace.

Resources you used:

  • The local python-task-race-lab folder containing myscript.py.

Keep everything running

No action needed. Choose this option if you want to rerun the benchmarks or extend the selector.

  • Keep the python-task-race-lab folder on your Mac.
  • Return to the integrated terminal whenever you want to repeat the lab.

Pause - I'll come back to this later

The script exits after the test report. No server or worker pool keeps running in the background.

  • Leave the python-task-race-lab folder in place.
  • Close the integrated terminal when you finish this session.

Delete - I don't want to use this again

Deleting the workspace is permanent. The command below removes the lab folder with myscript.py inside it.

  • Copy myscript.py somewhere outside python-task-race-lab first if you want to keep the experiment.
  • In the integrated terminal from earlier, remove the lab from its parent folder by running these commands:
cd ..
rm -rf python-task-race-lab
ls

What do these commands do?

  • The cd .. command moves the terminal out of the workspace before deletion.
  • The rm -rf python-task-race-lab command removes the project folder with its contents.
  • The ls command lists the folders that remain so you can confirm the cleanup.

You should no longer see python-task-race-lab in the output.

Still see the folder?

Check that the first command moved you to the folder containing python-task-race-lab.

Use this if you want help without guessing at another path: Help me safely locate and delete my project folder..

Nice Work!

Nice Work!

You did it! Your tested Python concurrency lab now compares three execution strategies. Its evidence shows which strategy fits each workload.

You've learned how to:

  • Measure sequential execution for waiting tasks. Repeat the baseline for CPU-heavy prime checks.
  • Reuse one task function with a thread pool to overlap waiting work. Use returned durations as the correctness check. Treat elapsed time as an observation.
  • Trigger a real pickling failure at the process boundary. Fix it with a module-scope worker protected by a process-safe entry point. Prove equivalent results with unit tests that check child process IDs.
  • Complete the Secret Mission by adding a tested executor selector. It maps I/O-bound work to threads. It maps CPU-bound work to processes.

Ready to quiz yourself?