Race Conditions

from pprint import pprint
from concurrent.futures import ThreadPoolExecutor
import threading

counter = 0
thread_iter = dict()

def change_counter(amount):
    global counter
    # to keep track of all the threads and total iterations each one is doing
    # along with the amount value that thread has been assigned
    if (threading.get_ident(), amount) not in thread_iter:
        thread_iter[(threading.get_ident(), amount)] = 0

    for _ in range(10000):
        thread_iter[(threading.get_ident(), amount)] += 1
        counter += amount

def race(num_threads):
    global counter
    counter = 0
    data = [-1 if x % 2 else 1 for x in range(1000)]

    # when the num_threads value is 1, the program is synchronous
    with ThreadPoolExecutor(max_workers=num_threads) as executor:
        executor.map(change_counter, data)

    pprint(thread_iter)
    print(f'counter value(ideally should be 0): {counter}')
    print(f'total iterations (1,000 x 10,000): {sum(thread_iter.values())}')

    delta = 0
    for k, v in thread_iter.items():
        delta += v if k[1] > 0 else -v
    print(f'delta (positive iterations - negative iterations): {delta}')

if __name__ == '__main__':
    # race(1)
    race(5)  # increase num_threads to consistently achieve race condition

"""
OUTPUT: 
{(1272, -1): 960000,
 (1272, 1): 1020000,
 (5704, -1): 940000,
 (5704, 1): 930000,
 (7960, -1): 1000000,
 (7960, 1): 1050000,
 (12428, -1): 1000000,
 (12428, 1): 900000,
 (12800, -1): 1100000,
 (12800, 1): 1100000}
counter value(ideally should be 0): 111490
total iterations (1,000 x 10,000): 10000000
delta (positive iterations - negative iterations): 0
"""

I tried to keep track of each thread, what amount it is using and how many iterations it is doing with that amount.

From the above output, it is clearly visible that each thread is doing different number of iterations for different amount value, but if we take the delta for iterations done for amount = 1 and amount = -1, it comes out as zero. Yet the counter value comes out as some arbitrary number instead of 0.

I am not able to make sense of this output :(

Also I am finding it difficult to walk through the program and understand exactly how the race condition is occurring.

Hi hauntarl,

The race condition is a caused by a combination of two things: the variable counter being global and shared across all threads, and the fact that the change_counter() method can be interrupted at any time.

The crux of the matter is the assumption that “counter += amount” can’t be interrupted. If you put a semaphore lock around “counter += amount” the code would likely work fine.

I suspect what is happening is that “counter += amount” breaks down into several different instructions in the underlying virtual machine and occasionally it is being interrupted midway through those instructions.

The code you added to try to show the problem happening is only recording the state outside of the problem area: after the increment/decrement is done. To truly instrument the problem you would have to be able to inspect the value of “counter” during the “counter += amount” statement.

The reason I used two large ranges (once in the “data” variable, the second in the for-loop in change_counter() ) was to increase the likelihood of just such an interruption. If you drop those numbers down to lower values you might have to run the program a large number of times before you run into a problem – this is what makes race conditions so brutal to catch in real life.

Thanks for the wonderful explanation, I think I finally get it.

If we try to breakdown the statement counter += amount in a 2 threaded system:

• evaluation of expression counter + amount: (Read operation)

  thread1: counter = 0, amount = 1, evaluation results in 1

  thread2: counter = 0, amount = -1, evaluation results in -1

As there is no mutual exclusion, both the threads simultaneously read the same “counter” and are trying to add “amount” to it.

This here is a race condition as no thread is willing to wait for the other one to update the counter value after its evaluation, which results in one of them using the old counter value depending on which thread is able to perform write operation first.

• assignment operation counter = evaluated value: (Write operation)

  thread1: counter = 1

  thread2: counter = -1

Let us assume that “thread1” is able to perform the write operation before “thread2”, the new “counter” value becomes 1, soon after “thread2” performs its own write operation and writes -1 to the “counter” Finally the counter value results in -1, instead of 0 :)

Correct me if I am wrong and feel free to provide more input.

Hi hauntarl,

Yep, you’ve got the right idea. The specifics are slightly different, but conceptually you’ve got it.

Remember that Python is an interpreted language. Statements in Python get translated into a series of steps called byte code, it is the byte code that actually gets run by the interpreter. Why do I bring this up? Because even things that look like single statements in Python aren’t necessarily atomic.

You can see the bytes that are run by the interpreter using the “compile()” built-in and the “dis” library:

>>> counter = 0
>>> amount = 1
>>> code = compile("counter += amount", "<string>", "exec")
>>> code
<code object <module> at 0x7fe81635ec90, file "<string>", line 1>
>>> import dis
>>> dis.disco(code)
  1           0 LOAD_NAME                0 (counter)
              2 LOAD_NAME                1 (amount)
              4 INPLACE_ADD
              6 STORE_NAME               0 (counter)
              8 LOAD_CONST               0 (None)
             10 RETURN_VALUE

Notice here that “counter += amount” turns into 6 separate byte code operations, and the thread interruption can happen during any of these steps.

The problem is the same as you described, just worse – there are five different places that this thing that looks like a single statement can get interrupted. This only gets worse for any larger chunk of code.

“ The sum total of the multiplication factors is not the same as the synchronous addition, so you get a race condition. ” This doesn’t intuitively explain why the race condition occurs as already pointed out in earlier comments . Essentially :

1. Multiple threads could be reading the same state of a particular variable ( in this case counter ) .

2. Subsequently, out of the evaluations ( of the expression counter + amount ) being performed by multiple threads ( that would be active at that particular point in time ) , only one of the threads effectively ends up updating counter ( overwriting the value assigned by the other threads) . This in turn could lead to either a ‘deficit’ or ‘surplus’ value of counter depending on which threads ( the ones adding -1 v/s those adding +1 ) dominated the in ‘final’ assigment of counter in each group of currently active threads .

In the synchronous version, mathematical equilibrium is maintained because each evaluation of counter += amount forms the basis for the next evaluation of this statement . But , in the concurrent version , that doesn’t happen as multiple threads could be using the same ‘base’ value for starting their computation of counter + amount and as a result most of the iterations do not contribute towards maintaining the equilibrium of counter .

Of course , this is an oversimplified explanation and not accurate ( given the complexities already pointed out the author in the comments ) .

I think Python uses GIL so only one thread excutes at any one time. When GIL is applied, race condition should occur but it is… Why does a race condition occur? (Sorry for my English)

@hwoarang09 The Global Interpreter Lock (GIL) ensures that only one CPython instruction can work at any given time, effectively making it an atomic operation that can’t be interrupted by another thread. However, in many cases, you’ll want to perform a transaction comprised of more than one instruction, which should all run as an atomic operation. Without explicit locking and mutexes, you can still experience a race condition.

@hwoarang09 The Global Interpreter Lock (GIL) ensures that only one CPython instruction can work at any given time, effectively making it an atomic operation that can’t be interrupted by another thread. However, in many cases, you’ll want to perform a transaction comprised of more than one instruction, which should all run as an atomic operation. Without explicit locking and mutexes, you can still experience a race condition.

However, the situation can be easily observed if the counter is stored or cached locally in the context of specific thread and update only at the end of computations. Also, the bigger values of delay is simulating, the bigger chance of encountering corrupted data.

Here are some modifications of the code

Example from this video just only with counters cached locally (usage of “local_counter”)

# io_bound/race.py
from concurrent.futures import ThreadPoolExecutor
counter = 0

def change_counter(amount):
    global counter
    local_counter = counter
    for _ in range(10000):
        local_counter += amount

    counter = local_counter

def race(num_threads):
    global counter
    counter = 0
    data = [-1 if x %2 else 1 for x in range(1000)]

    with ThreadPoolExecutor(max_workers=num_threads) as executor:
        executor.map(change_counter, data)

    print(counter)

More advanced example for both “manual” and “ThreadPool-way” of creating and running threads; also simulating delay as a parameter

import argparse
import threading
import random
import time

from typing import List, Tuple
from concurrent.futures import ThreadPoolExecutor


counter = 0

def change_counter(data: Tuple[List[int], bool]):
    global counter

    for d in data[0]:
        local_counter = counter
        local_counter += d

        print(data[0], data[1])
        if data[1]:
            time.sleep(random.random())

        counter = local_counter

        print(f'{counter=} ({threading.current_thread().name})')


def prepare_data(num_threads: int) -> List[List[int]]:
    """
    Gets sample data for threads. Each thread has its own array of three integers multiplied by index + 1.
    Example:
        num_threads = 2, data = [[1, 2, 3], [2, 4, 6]]
        num_threads = 3, data = [[1, 2, 3], [2, 4, 6], [3, 6, 9]]
    :param num_threads: number of threads
    :return: array of 3-number arrays for each thread
    """
    data = [[]] * num_threads

    for n in range(num_threads):
        data[n] = [(i + 1) * (n + 1) for i in range(3)]

    print(data)
    return data

def race_threads_manually(num_threads: int, simulate_delay: bool):
    data = prepare_data(num_threads)

    # Creating and running threads
    T = [None] * num_threads

    for it in range(len(T)):
        T[it] = threading.Thread(target=change_counter, args=([(data[it], simulate_delay)]), name=f'Thread {it + 1}')
        T[it].start()

    # Waiting for all to finish
    for it in range(len(T)):
        T[it].join()

    print(f'{counter=}')

def race_thread_pool(num_threads: int, simulate_delay: bool):
    data = prepare_data(num_threads)
    args = [(data[d], simulate_delay) for d in range(num_threads)]

    with ThreadPoolExecutor(max_workers=num_threads) as executor:
        executor.map(change_counter, args)

    print(f'{counter=}')

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('-n', '--threads_number', type=int, help='Number of threads.', default=1)
    parser.add_argument('-t', '--simulation_type', choices=['loop', 'pool'], help='The way of running threads: by thread pool or in a loop.', default='pool')
    parser.add_argument('-d', '--simulate_delay', action='store_true', help='Simulates delay in milliseconds.', default=False)
    args = parser.parse_args()

    if args.simulation_type == 'pool':
        func = race_thread_pool
    else:
        func = race_threads_manually

    func(args.threads_number, args.simulate_delay)