Synchronization with ThreadPoolExecutor in Python

Question

I have written a program for executing my tasks parallelly with ThreadPoolExecutor and the method which I am calling to execute parallelly, giving errors, probably bcoz of the race condition. So I want to avoid the race condition for particular portion of the method. Below is my current code:

def my_multithreading_method(my_list):
    my_list.append("Hello")
    return True

def parallel_execution(my_list):
    try:
        response_list = []
        future_list =[]
        list_count = len(my_list)
        with ThreadPoolExecutor(max_workers=list_count) as executor:
            for index in range(list_count):
                future_list.append(executor.submit(my_multithreading_method, my_list[index]))
            for future in as_completed(future_list):
                return_value = future.result()
                response_list.append(return_value[0])
        return response_list
    except Exception as error:
        raise Exception(str(error))

For example I have to avoid the race the condition inside my_multithreading_method for the operation my_list.append(), what I shall I do? It can be any operation like file operation. Along with this operation, there are many other operations, taking place before this operation inside the function. Please help me with some efficient method.

Answer 1

You can use the Lock object to make a section of code thread safe. Below is an example how you can add a lock to protect the line from my_multithreading_method:

from concurrent.futures import ThreadPoolExecutor, as_completed
from threading import Lock


def my_multithreading_method(my_list, lock):
    with lock:
        my_list.append("Hello")
        return True

lock = Lock()

def parallel_execution(my_list):
    try:
        response_list = []
        future_list = []
        list_count = len(my_list)
        with ThreadPoolExecutor(max_workers=list_count) as executor:
            for index in range(list_count):
                future_list.append(executor.submit(my_multithreading_method, my_list[index], lock))
            for future in as_completed(future_list):
                return_value = future.result()
                response_list.append(return_value[0])
        return response_list
    except Exception as error:
        raise Exception(str(error))