Я попадаю в тупик, когда пытаюсь использовать многопроцессорность с pytorch. Эквивалентный кодовый код работает так, как я ожидал.
Я сделал упрощенную версию своего кода: пул из 4 рабочих, выполняющих широковещательную операцию в массиве 1000 раз (т.е. ~ 250 на каждого работника). Рассматриваемый массив имеет размер 100 000 x 3, и операция широковещания заключается в вычитании всех строк из одного массива 1 x 3 строки. Большой массив является общим / глобальным массивом, и массив строк отличается на каждой итерации.
Код работает точно так же, как и ожидалось, используя numpy, при этом рабочие в пуле демонстрируют 4-кратное ускорение по сравнению с циклом for.
Однако код в pytorch заходит в тупик (я полагаю): ни один из рабочих не завершает операцию широковещательной передачи массива ни разу.
Код Numpy ниже печатает следующее:
Закончено для цикла над my_subtractor: прошло 8,1504 секунды.
Законченный пул через my_subtractor: заняло 2,2247 секунды.
Код pytorch, с другой стороны, печатает это, затем останавливается:
Закончено для цикла по my_subtractor: заняло 3,1082 секунды.
BLA
BLA
BLA
BLA
Печатные операторы "BLA" просто показывают, что каждый работник застрял в - по-видимому - тупиковом состоянии. Их ровно 4: по одному на каждого работника, входящего - и застревающего - итерация.
Если вы чувствуете себя достаточно амбициозным для воспроизведения, обратите внимание, что он не работает в Windows, потому что он не обернут вокруг if __name__ == '__main__': (я где-то читал, что вам это нужно из-за способа, которым Windows обрабатывает процессы запуска). Также вам нужно будет создать пустой файл с именем my_globals.py.
Вот код NumPy
from time import time
import numpy as np
import my_globals
from multiprocessing import Pool as ThreadPool
# shared memory by virtue of being global
my_globals.minuend = np.random.rand(100000,3)
# array to be iterated over in for loop / pool of workers
subtrahends = np.random.rand(10000,3)
# function called at each iteration (broadcast operation)
def my_subtractor(subtrahend):
my_globals.minuend - subtrahend
return 0
# launch for loop
ts = time()
for idx, subtrahend in enumerate(subtrahends):
my_subtractor(subtrahend)
te = time()
print('Finished for loop over my_subtractor: took %2.4f seconds.' % (te - ts))
# launch equivalent pool of workers
ts = time()
pool = ThreadPool(4)
pool.map(my_subtractor, subtrahends)
pool.close()
pool.join()
te = time()
print('Finished pool over my_subtractor: took %2.4f seconds.' % (te - ts))
Вот эквивалентный код pytorch:
from time import time
import torch
import my_globals
from torch.multiprocessing import Pool as ThreadPool
# necessary on my system because it has low limits for number of file descriptors; not recommended for most systems,
# see: https://pytorch.org/docs/stable/multiprocessing.html#file-descriptor-file-descriptor
torch.multiprocessing.set_sharing_strategy('file_system')
# shared memory by virtue of being global
my_globals.minuend = torch.rand(100000,3)
# array to be iterated over in for loop / pool of workers
subtrahends = torch.rand(10000,3)
# function called at each iteration (broadcast operation)
def my_subtractor(subtrahend, verbose=True):
if verbose:
print("BLA") # -- prints for every worker in the pool (so 4 times total)
my_globals.minuend - subtrahend
if verbose:
print("ALB") # -- doesn't print for any worker
return 0
# launch for loop
ts = time()
for idx, subtrahend in enumerate(subtrahends):
my_subtractor(subtrahend, verbose=False)
te = time()
print('Finished for loop over my_subtractor: took %2.4f seconds.' % (te - ts))
# launch equivalent pool of workers
ts = time()
pool = ThreadPool(4)
pool.map(my_subtractor, subtrahends)
pool.close()
pool.join()
te = time()
print('Finished pool over my_subtractor: took %2.4f seconds.' % (te - ts))