Common Python Concurrency Mistakes

Every concurrency bug in Python looks correct when you read the code. The problem is not in the logic you wrote but in the assumptions you brought from single-threaded programming. A variable that increments correctly in one thread gets silently corrupted in two.

A function that returns promptly in synchronous code blocks the entire event loop in async code. A list modification that works in a single process crashes or produces garbage when shared across processes. This article catalogs the most common concurrency mistakes across threading, asyncio, and multiprocessing, with the specific fix for each.

If you have worked through the decision guide for choosing a concurrency model, you know which tool fits your workload. This article helps you avoid the mistakes that make even the right tool produce wrong results.

The mistakes are organized by category: GIL misunderstandings, synchronization failures, event loop blocking, task lifecycle bugs, and process isolation errors. Each section shows a buggy example, explains why it fails, and gives the corrected version.

Mistake 1: Using threads for CPU-bound work

Reaching for threads to speed up a computation-heavy function is a natural instinct from other languages, but in Python it does not work the way you would expect:

python

# BUG: Threading cannot parallelize CPU work
import threading
 
def count_primes(limit):
    count = 0
    for n in range(2, limit + 1):
        for d in range(2, int(n ** 0.5) + 1):
            if n % d == 0:
                break
        else:
            count += 1
    return count
 
threads = [
    threading.Thread(target=count_primes, args=(50000,))
    for _ in range(4)
]
for t in threads: t.start()
for t in threads: t.join()

This runs slower than calling count_primes four times sequentially because the GIL allows only one thread to execute Python bytecode at a time. The thread switching overhead adds cost with no parallelism benefit.

Fix: Use multiprocessing for CPU-bound work. Each worker process gets its own interpreter and its own GIL, so the four prime-counting calls actually run on separate cores instead of taking turns on one:

python

from concurrent.futures import ProcessPoolExecutor
 
if __name__ == "__main__":
    with ProcessPoolExecutor() as executor:
        results = list(executor.map(count_primes, [50000] * 4))

Mistake 2: Assuming the GIL makes all code thread-safe

The GIL prevents two threads from executing Python bytecode simultaneously, but it does not make compound operations atomic. counter += 1 is three bytecode operations, and the GIL can switch threads between any two of them.

python

# BUG: counter += 1 is not atomic
import threading
 
counter = 0
 
def increment():
    global counter
    for _ in range(1_000_000):
        counter += 1
 
t1 = threading.Thread(target=increment)
t2 = threading.Thread(target=increment)
t1.start(); t2.start()
t1.join(); t2.join()
 
print(counter)  # Less than 2_000_000

Fix: Use a lock or a thread-safe data structure. Wrapping the increment in with lock: forces the read-increment-write sequence to complete in one thread before another thread can start it:

python

import threading
 
counter = 0
lock = threading.Lock()
 
def increment():
    global counter
    for _ in range(1_000_000):
        with lock:
            counter += 1

Mistake 3: Forgetting to join threads

Starting a thread and moving on without waiting for it leaves you with no guarantee about what has actually finished by the time your program continues or exits:

python

# BUG: The main thread may exit before worker finishes
import threading
import time
 
def slow_work():
    time.sleep(2)
    print("Work done")
 
t = threading.Thread(target=slow_work)
t.start()
print("Main exiting")  # May print and exit before "Work done"

If the worker thread is a daemon, it is killed silently. If it is non-daemon, the program waits, but you have no way to know when it finishes or collect its result.

Fix: Always join() threads whose results you need, so the main thread waits for the worker to actually finish before it moves on:

python

t = threading.Thread(target=slow_work)
t.start()
t.join()
print("Main exiting")  # Prints after "Work done"

Mistake 4: Creating tasks without awaiting them

Calling a coroutine function only builds a coroutine object; nothing inside it runs until something awaits it or schedules it on the event loop:

python

import asyncio
 
async def fetch(url):
    await asyncio.sleep(0.1)
    return f"Content of {url}"
 
async def main():
    # BUG: Coroutine created but never awaited
    fetch("https://example.com")  # RuntimeWarning: coroutine was never awaited
 
asyncio.run(main())

The coroutine is never scheduled and its result is lost. Python issues a RuntimeWarning when the coroutine is garbage collected without ever being awaited, but by then the work it should have done never happened.

Fix: Either await the coroutine directly or wrap it in a task and await the task later, so the event loop actually schedules and runs it instead of discarding it unused:

python

async def main():
    result = await fetch("https://example.com")
    print(result)

Mistake 5: Blocking the asyncio event loop

An async def function can still call ordinary blocking code, and Python will not stop you, even though doing so defeats the entire point of using asyncio:

python

import asyncio
import time
 
async def io_task():
    await asyncio.sleep(0.1)
    return "done"
 
async def main():
    # BUG: time.sleep blocks the entire event loop
    time.sleep(2)
    result = await io_task()

time.sleep() is synchronous. It blocks the thread, which blocks the event loop, which blocks every other async task. The io_task cannot make progress until the sleep completes.

This is the single most common asyncio mistake.

Fix: Use asyncio.sleep() in async code instead of time.sleep(), or wrap unavoidable synchronous calls in run_in_executor() so they run in a thread pool instead of on the event loop itself:

python

async def main():
    await asyncio.sleep(2)
    result = await io_task()

Mistake 6: Using synchronous HTTP clients in async code

This is a specific, very common case of the previous mistake: reaching for a familiar synchronous library inside async code without realizing it blocks the whole event loop:

python

import asyncio
import requests
 
async def fetch(url):
    # BUG: requests.get blocks the event loop
    response = requests.get(url)
    return response.text
 
async def main():
    results = await asyncio.gather(
        fetch("https://a.example.com"),
        fetch("https://b.example.com"),
    )

requests.get() is synchronous. Even though it is called inside an async def, it blocks the event loop. The second fetch does not start until the first completes.

The whole program runs sequentially.

Fix: Use an async HTTP client like aiohttp or httpx with async support, so the request itself yields control at its own await point instead of blocking the whole event loop:

python

import aiohttp
 
async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()

Mistake 7: Deadlock from inconsistent lock ordering

When two threads each need two locks but grab them in a different order, both can end up waiting on each other forever:

python

import threading
 
lock_a = threading.Lock()
lock_b = threading.Lock()
 
def thread_1():
    with lock_a:
        with lock_b:
            pass
 
def thread_2():
    with lock_b:     # Acquired in opposite order
        with lock_a:
            pass

If thread 1 holds lock_a and thread 2 holds lock_b simultaneously, both threads block forever. This is a classic deadlock.

Fix: Always acquire locks in the same order across every thread, so no thread can ever hold one lock while waiting on a lock another thread already holds:

python

def thread_2():
    with lock_a:     # Same order as thread_1
        with lock_b:
            pass

Mistake 8: Holding a lock during I/O

A lock only needs to protect the shared state it guards, not the entire operation that happens to touch that state:

python

import threading
 
lock = threading.Lock()
results = []
 
def fetch_and_store(url):
    with lock:
        data = requests.get(url).text  # I/O while holding lock
        results.append(data)

The lock is held for the entire duration of the HTTP request. Every other thread blocks waiting for the lock even though they could be doing their own I/O concurrently.

Fix: Move I/O outside the locked section, and only hold the lock long enough to append the already-fetched data to the shared list:

python

def fetch_and_store(url):
    data = requests.get(url).text     # I/O without lock
    with lock:
        results.append(data)

Mistake 9: Forgetting task_done with queues

queue.Queue tracks outstanding work through task_done() calls, and it has no other way of knowing that a get()'d item has actually finished processing:

python

import queue
import threading
 
q = queue.Queue()
 
def consumer():
    while True:
        item = q.get()
        if item is None:
            break
        process(item)
        # BUG: task_done() never called
 
# Producer calls q.join() and blocks forever

q.join() blocks until every get() has been matched by a task_done(). If a consumer forgets task_done(), the producer blocks forever.

Fix: Always call task_done() after processing each item, including the sentinel value, so the count of completed items eventually matches the count of items put on the queue:

python

def consumer():
    while True:
        item = q.get()
        if item is None:
            q.task_done()  # Match the sentinel too
            break
        try:
            process(item)
        finally:
            q.task_done()

Mistake 10: Sharing mutable state across processes

Code that works fine with threads sharing a variable often breaks silently when the same variable is shared across processes, since processes do not share memory the way threads do:

python

import multiprocessing
 
data = []
 
def worker():
    data.append("modified")
    print(f"Worker sees: {data}")
 
if __name__ == "__main__":
    p = multiprocessing.Process(target=worker)
    p.start()
    p.join()
    print(f"Parent sees: {data}")  # Still empty

Processes do not share memory. The worker appends to its own copy of data. The parent's data is unchanged.

Fix: Use multiprocessing.Manager, Queue, or Value/Array for sharing data between processes, since a plain global variable is never actually shared once a second process exists:

python

import multiprocessing
 
def worker(shared_list):
    shared_list.append("modified")
 
if __name__ == "__main__":
    with multiprocessing.Manager() as manager:
        shared = manager.list()
        p = multiprocessing.Process(target=worker, args=(shared,))
        p.start()
        p.join()
        print(f"Parent sees: {list(shared)}")

Mistake 11: Using the fork start method with threads

If you use the fork start method on Linux and the parent process has running threads, the child process inherits a copy of the parent's memory but only the forking thread is active. Any locks held by other threads in the parent are locked forever in the child. This causes mysterious deadlocks.

Fix: Use spawn or forkserver in multithreaded programs, or ensure no threads are running when you fork.

python

import multiprocessing
 
multiprocessing.set_start_method("spawn")

Mistake 12: Neglecting exception handling in concurrent code

A task that raises an exception does not crash your program the way an uncaught exception in synchronous code would; it just fails quietly unless something checks on it:

python

import asyncio
 
async def worker():
    await asyncio.sleep(0.1)
    raise ValueError("Silent failure")
 
async def main():
    task = asyncio.create_task(worker())
    await asyncio.sleep(0.2)  # Task already failed
    # BUG: Never checked task.exception()

The task raises an exception, but main() never awaits it or checks task.exception(). The failure is silent.

Fix: Always await tasks or check their exception state.

python

async def main():
    task = asyncio.create_task(worker())
    try:
        await task
    except ValueError as e:
        print(f"Task failed: {e}")

Mistake checklist

Before deploying concurrent Python code, verify:

1. Threads are used for I/O, not CPU. Multiprocessing handles CPU-bound work.

2. Every shared mutable variable is protected by a lock, queue, or other synchronization primitive.

3. +=, dictionary updates, and list operations on shared data are inside locked sections.

4. No synchronous I/O or time.sleep() exists inside async functions.

5. Every create_task() has a corresponding await, TaskGroup, or explicit exception check.

6. Locks are acquired in a consistent order across all code paths.

7. Locks are not held during I/O or long computations.

8. task_done() is called for every get() on a queue.

9. Inter-process communication uses Queue, Manager, Value, or shared_memory, not global variables.

10. Exception handlers exist in every thread and task, and failures are surfaced to the main execution path.