Parallel map

[thedeemon]

Не раз уже видел восторженные отчеты об успешном применении Эрланга в задаче скачивания большого числа файлов. На мое замечание, что распараллеливание кучи одинаковых независимых заданий - невелико достижение, мне ответили, что для обычных императивных языков это вполне достижение. Сегодня я сам столкнулся с такой задачей, и на совершенно императивном языке Руби задача решилась очень просто.


Недавно узнал о сайте с комиксами abstrusegoose.com, местами попадаются неплохие, но листать по одному задалбывает. Решил сгенерить себе страничку со всеми сериями сразу.

Функция, которая скачивает страничку очередной серии, выдирает из нее нужный кусочек и приводит его к нужному виду, выглядит так:

require 'open-uri'

def fetch(n, tries=5)
  print n, "..\n"
  begin
    data = open("http://abstrusegoose.com/#{n}").read
  rescue
    if tries > 0 then return fetch(n, tries-1)
    else data = "" end
  end
  r = %r{<p><img class="aligncenter".*?</p>}m
  "#{n}<br> #{r.match(data).to_s} <hr>\n"
end

Получить последовательно все кусочки и собрать в одну строку можно так:

all = ""
for n in 1..267
  all << fetch(n)
end

Но это не слишком быстро, большая часть времени проходит в ожидании прихода данных. Зато такое ожидание - как раз одна из немногих вещей, которые в Руби поддаются эффективному распараллеливанию. :) Добавим в класс массивов операцию параллельного отображения (map):

class Array
 def par_map(nt, &f)
  slices = Array.new(nt) {|i| self[i*length/nt...(i+1)*length/nt] }
  threads = slices.map {|xs| Thread.new { xs.map &f } }
  threads.inject([]) {|r, t| r + t.value}
 end
end

Параметр nt - число потоков, на которое распараллеливать.
В первой строчке исходный массив делится на примерно равные части, количество которых равно числу потоков.
Во второй строчке для каждой части создается поток, который применяет переданную в метод операцию ко всем элементам своей части. Все созданные потоки сохраняются в массиве threads. При покидании второй строчки потоки созданы, некоторые, возможно, уже начали работу.
В третьей строчке результаты работы потоков объединяются в один массив. Для получения результата работы потока t идет обращение к свойству t.value, которое дожидается окончания его работы, прежде чем вернуть результат. inject - это обычный fold так в Руби называется.
Вот и все, три строчки - все распараллеливание.

Запустить скачивание в 10 потоков и собрать результаты в одну строку можно так:

all = (1..267).to_a.par_map(10) {|n| fetch(n)}.join

Тут range 1..267 превращается в массив со значениями от 1 до 267, у него вызывается только что описанный метод параллельного отображения, которому передается число потоков и операция для элемента. Результирующий массив строк собирается в одну методом join. Задача решена! Ускорение в данном случае близко к линейному.

Из-за GIL распараллелить таким образом вычисления на Руби не выйдет, зато благодаря GIL нет никаких проблем с синхронизацией.

Comments

[zeux.livejournal.com]

Поправьте меня, если я ошибаюсь, но...
- в данном примере наличие или отсутствие GIL ничего не изменило бы
- в общем случае GIL совсем не отменяет синхронизации из-за сложных состояний, меняющихся не атомарно (больше чем за 1 такт с т.з. GIL).

[thedeemon.livejournal.com]

1. Да, здесь бы не изменило. Но если бы не было GIL, можно было бы использовать этот метод для паралеллизации вычислений.
2. Да, сложные изменения надо синхронизировать.