Threads не выполняется параллельно python с ThreadPoolExecutor - PullRequest
Новая
       23

Threads не выполняется параллельно python с ThreadPoolExecutor

1 голос
/ 11 апреля 2020

im new в python потоке, когда я печатаю вывод, кажется, что он работает параллельно. Также мои функции занимают то же время, что и перед использованием библиотеки concurrent.futures (ThreadPoolExecutor). Я должен рассчитать прирост некоторых атрибутов над набором данных (я не могу использовать библиотеки). Поскольку у меня есть около 1024 атрибутов, а выполнение функции заняло около минуты (и я должен использовать ее для итерации), я решил разделить массив attributes на 10 (просто в качестве примера) и запустить separete. Функция gain(attribute) отдельно для каждого подмассива. Итак, я сделал следующее (избегая лишних ненужных кодов): 

def calculate_gains(self):
    splited_attributes = np.array_split(self.attributes, 10)
    result = {}
    for atts in splited_attributes:
        with concurrent.futures.ThreadPoolExecutor() as executor:
            future = executor.submit(self.calculate_gains_helper, atts)
            return_value = future.result()
            self.gains = {**self.gains, **return_value}

Вот вот посчитанный_параметром: 

def calculate_gains_helper(self, attributes):
    inter_result = {}
    for attribute in attributes:
        inter_result[attribute] = self.gain(attribute)
    return inter_result

Я что-то не так делаю? Я прочитал некоторые другие старые сообщения, но я не мог получить какую-либо информацию. Большое спасибо за любую помощь!

Ответы [ 2 ]

1 голос
/ 11 апреля 2020
Потоки

Python не работают параллельно (по крайней мере, в реализации CPython) из-за GIL . Используйте процессы и ProcessPoolExecutor , чтобы действительно иметь параллелизм 

with concurrent.futures.ProcessPoolExecutor() as executor:
    ...
0 голосов
/ 11 апреля 2020

Вы отправляете и затем последовательно ожидаете каждого рабочего элемента, поэтому все потоки замедляют работу. Я не могу гарантировать, что это сильно ускорит процесс, потому что вы все еще имеете дело с python GIL, который не позволяет вещам уровня python работать параллельно, но здесь все идет.

Я создал поток объединить все возможности в рабочий, включая нарезку self.attributes. 

def calculate_gains(self):
    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        result_list = executor.map(self.calculate_gains_helper,
            ((i, i+10) for i in range(0, len(self.attributes), 10)))
    for return_value in result_list:
        self.gains = {**self.gains, **return_value}

def calculate_gains_helper(self, start_end):
    start, end = start_end
    inter_result = {}
    for attribute in self.attributes[start:end]:
        inter_result[attribute] = self.gain(attribute)
    return inter_result
...