У меня есть многопроцессорное задание, в котором я выполняю очередь с использованием только массивов numpy, как часть конвейера производителя.
В настоящее время они маринуются, потому что это поведение по умолчанию multiprocessing.Queue, которое замедляет производительность.
Есть ли какой-нибудь pythonic способ передать ссылки на разделяемую память вместо того, чтобы травить массивы?
К сожалению, массивы генерируются после запуска потребителя, и нет простого способа обойти это. (Таким образом, глобальный переменный подход был бы уродливым...).
[Обратите внимание, что в следующем коде мы не ожидаем вычисления h (x0) и h (x1) параллельно. Вместо этого мы видим, что h (x0) и g (h (x1)) вычисляются параллельно (например, конвейерная обработка в ЦП).]
from multiprocessing import Process, Queue
import numpy as np
class __EndToken(object):
pass
def parrallel_pipeline(buffer_size=50):
def parrallel_pipeline_with_args(f):
def consumer(xs, q):
for x in xs:
q.put(x)
q.put(__EndToken())
def parallel_generator(f_xs):
q = Queue(buffer_size)
consumer_process = Process(target=consumer,args=(f_xs,q,))
consumer_process.start()
while True:
x = q.get()
if isinstance(x, __EndToken):
break
yield x
def f_wrapper(xs):
return parallel_generator(f(xs))
return f_wrapper
return parrallel_pipeline_with_args
@parrallel_pipeline(3)
def f(xs):
for x in xs:
yield x + 1.0
@parrallel_pipeline(3)
def g(xs):
for x in xs:
yield x * 3
@parrallel_pipeline(3)
def h(xs):
for x in xs:
yield x * x
def xs():
for i in range(1000):
yield np.random.uniform(0,1,(500,2000))
if __name__ == "__main__":
rs = f(g(h(xs())))
for r in rs:
print r
