Параллельный Питон, начало

Disclaimer

Родилась у товарища географическая потребность перенести кусочек карты из одного участка Земли в другой. Он это по привычке сделал на дельфях, мне же захотелось попробовать в действии питон, в коем я спецом не являюсь.

Практика

Собственно перевести алгоритм оказалось делом совсем несложным, но вот скорость его работы оставляла желать лучшего.
Первым делом в ход пошел Psyco, ускорив обработку в 6 раз.

Получить лучший результат без изменения алгоритма уже не представлялось возможным, поэтому в ход пошел метод грубой силы — распараллеливание задач.

Найден был модуль Parallel Python. Подключить его оказалось делом совсем несложным:

Сначала import pp, а потом (первый вариант):

ppservers = ()
job_server = pp.Server(ppservers=ppservers)                     

job_server.set_ncpus(2)
print "Starting pp with", job_server.get_ncpus(), "workers"

jobs = [job_server.submit(tighina_check, (), (find_geo_coords, compare, get_dist_bearing,), ("math", ))  for i in range(3)]

for job in jobs:
	job()

job_server.print_stats()

Код, в принципе, сам за себя говорящий — используем только локальный сервер (а вообще модуль позволяет распараллеливать и на сетевые), стараемся запустить на 2-х процессорах, указываем какую функцию вызывать и от каких она зависит, импортируем math и запускаем 3 задачи, в конце печатаем статистику.

Первой засадой оказалось отключение psyco, что опять отбросило нас на стартовую позицию.
Решение было очевидным — добавить импорт psyco при создании job'а
jobs = [job_server.submit(tighina_check, (), (find_geo_coords, compare, get_dist_bearing,), ("math", "psyco", )) for i in range(3)]
и вызывать psyco.full уже в tighina_check:

def tighina_check():
        psyco.full()
        #а вот тут много математики

Вторая проблема оказалась весьма неожиданной.
Код в tighina_check был изначально заточен под импорт вида «from math import sin, pow, cos, asin, sqrt, fabs, acos». Но он не работал под pp, т.к. создает среду выполнения функции только с модулями, указанными при создании job'а. Вполне логичным было переделать все вызовы sin на math.sin и т.д. Вот тут-то и возникло небольшое недоумение — интенсивное и постоянное использоваение мат.функций во втором формате вызова приводило к замедлению в 1.3-1.4 раза.

Решением было ручное импортирование нужных функций в глобальную область видимости в начале каждого job'a:

def tighina_check():
     psyco.full()
     math_func_imports = ['sin', 'cos', 'asin', 'acos', 'sqrt', 'pow', 'fabs']
     for func in math_func_imports:
	 setattr(__builtins__, func, getattr(math, func))

Дальше подумалось, что неплохо бы ускорить сам pp с помощью psyco. Для этого нужно немного подпатчить pyworker.py из комплекта, добавив в начало:

import psyco
psyco.full()

и заменив

eval(__fobj)

на

exec __fobj

При этом отпадает необходимость в импорте psyco при создании job'а и соответсвенно в вызове psyco.full() в job'e.

Остальное — только подборка нужного числа процессоров

Что в итоге?

Запускалось 100 job'ов.

Исходный вариант (никакого распараллеливания, только psycho)
100 последовательных job'ов 257 секунд

2 процессора (pp, psyco)

Starting pp with 2 workers
Job execution statistics:
 job count | % of all jobs | job time sum | time per job | job server
       100 |        100.00 |     389.8933 |     3.898933 | local
Time elapsed since server creation 195.12789011

4 процессора (pp, psyco)

Starting pp with 4 workers
Job execution statistics:
 job count | % of all jobs | job time sum | time per job | job server
       100 |        100.00 |     592.9463 |     5.929463 | local
Time elapsed since server creation 148.77167201

Дальше тестировать не хотелось, казалось, что 2 ядра, каждое с гипертредингом, а значит 4 job'а — оптимальный вариант. Но любопытство взяло вверх (и как оказалось — не зря):
8 процессоров (pp, psyco)

Starting pp with 8 workers
Job execution statistics:
 job count | % of all jobs | job time sum | time per job | job server
       100 |        100.00 |     1072.3920 |    10.723920 | local
Time elapsed since server creation 137.681350946

16 процессоров (pp, psyco)

Starting pp with 16 workers
Job execution statistics:
 job count | % of all jobs | job time sum | time per job | job server
       100 |        100.00 |     2050.8158 |    20.508158 | local
Time elapsed since server creation 133.345046043

32 процессора (pp, psyco)

Starting pp with 32 workers
Job execution statistics:
 job count | % of all jobs | job time sum | time per job | job server
       100 |        100.00 |     4123.8550 |    41.238550 | local
Time elapsed since server creation 136.022897005

Т.о. в лучшем варианте 133 секунды против 257 в первоначальном варианте = ускорение в 1.93 раза для нашей конкретной задачи только за счет распараллеливания.

Следует отметить, что все 100 job'ов друг от друга не зависят и не нуждаются в «общении» между собой, что облегчает задачу и увеличивает скорость.

Итоговые примеры кода:

ppservers = ()
job_server = pp.Server(ppservers=ppservers)                     

job_server.set_ncpus(16)
print "Starting pp with", job_server.get_ncpus(), "workers"

jobs = [job_server.submit(tighina_check, (), (find_geo_coords, compare, get_dist_bearing,), ("math", ))  for i in range(3)]

for job in jobs:
    job()

job_server.print_stats()

def tighina_check():
    math_func_imports = ['sin', 'cos', 'asin', 'acos', 'sqrt', 'pow', 'fabs']
    for func in math_func_imports:
        setattr(__builtins__, func, getattr(math, func)) 

        #а вот тут много математики